feat: Onboard Uniref50 dataset (#473)

GoogleCloudPlatform · Sep 13, 2022 · b44d572 · b44d572
1 parent de7f1fa
commit b44d572
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 29 deletions.
diff --git a/datasets/uniref50/pipelines/_images/run_csv_transform_kub/csv_transform.py b/datasets/uniref50/pipelines/_images/run_csv_transform_kub/csv_transform.py
@@ -234,7 +234,6 @@ def load_data_to_bq(
  job_config.write_disposition = "WRITE_TRUNCATE"
  else:
  job_config.write_disposition = "WRITE_APPEND"
- print("appending started")
  job_config.skip_leading_rows = 1 # ignore the header
  job_config.autodetect = False
  with open(file_path, "rb") as source_file:
@@ -331,9 +330,11 @@ def create_table_schema(
 def append_batch_file(target_file_batch: str, target_file: str) -> None:
 
  with open(target_file_batch, "r") as data_file:
- with open(target_file, "a+") as target_file:
+ with open(target_file, "a+") as _target_file:
  logging.info(f"Appending batch file {target_file_batch} to {target_file}")
- target_file.write(data_file.read())
+ logging.info(f"Size of target file is {os.path.getsize(target_file_batch)}")
+ logging.info(f"Size of target file is {os.path.getsize(target_file)}")
+ _target_file.write(data_file.read())
  if os.path.exists(target_file_batch):
  os.remove(target_file_batch)
 

diff --git a/datasets/uniref50/pipelines/uniref50/pipeline.yaml b/datasets/uniref50/pipelines/uniref50/pipeline.yaml
@@ -81,7 +81,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -151,7 +151,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -221,7 +221,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -291,7 +291,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -361,7 +361,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -431,7 +431,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -501,7 +501,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -571,7 +571,7 @@ dag:
  resources:
  request_memory: "4G"
  request_cpu: "1"
- request_ephemeral_storage: "50G"
+ request_ephemeral_storage: "10G"
 
  - operator: "GoogleCloudStorageToBigQueryOperator"
  description: "Task to load CSV data to a BigQuery table"
@@ -614,4 +614,4 @@ dag:
  description: ""
  mode: "NULLABLE"
  graph_paths:
- - "download_zip_file >> uniref50_transform_csv_1 >> load_uniref50_to_bq_1 >> uniref50_transform_csv_2 >> load_uniref50_to_bq_2 >> uniref50_transform_csv_3 >> load_uniref50_to_bq_3 >> uniref50_transform_csv_4 >> load_uniref50_to_bq_4 >> uniref50_transform_csv_5 >> load_uniref50_to_bq_5 >> uniref50_transform_csv_6 >> load_uniref50_to_bq_6 >> uniref50_transform_csv_7 >> load_uniref50_to_bq_7 >> uniref50_transform_csv_8 >> load_uniref50_to_bq_8"
+ - "download_zip_file >> [ uniref50_transform_csv_1, uniref50_transform_csv_2, uniref50_transform_csv_3, uniref50_transform_csv_4, uniref50_transform_csv_5, uniref50_transform_csv_6, uniref50_transform_csv_7, uniref50_transform_csv_8 ] >> load_uniref50_to_bq_1 >> load_uniref50_to_bq_2 >> load_uniref50_to_bq_3 >> load_uniref50_to_bq_4 >> load_uniref50_to_bq_5 >> load_uniref50_to_bq_6 >> load_uniref50_to_bq_7 >> load_uniref50_to_bq_8 "
diff --git a/datasets/uniref50/pipelines/uniref50/uniref50_dag.py b/datasets/uniref50/pipelines/uniref50/uniref50_dag.py
@@ -29,7 +29,7 @@
  dag_id="uniref50.uniref50",
  default_args=default_args,
  max_active_runs=1,
- schedule_interval="@weekly",
+ schedule_interval="@once",
  catchup=False,
  default_view="graph",
 ) as dag:
@@ -71,7 +71,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -143,7 +143,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -215,7 +215,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -287,7 +287,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -359,7 +359,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -431,7 +431,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -503,7 +503,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -575,7 +575,7 @@
  resources={
  "request_memory": "4G",
  "request_cpu": "1",
- "request_ephemeral_storage": "50G",
+ "request_ephemeral_storage": "10G",
  },
  )
 
@@ -622,20 +622,22 @@
 
  (
  download_zip_file
- >> uniref50_transform_csv_1
+ >> [
+ uniref50_transform_csv_1,
+ uniref50_transform_csv_2,
+ uniref50_transform_csv_3,
+ uniref50_transform_csv_4,
+ uniref50_transform_csv_5,
+ uniref50_transform_csv_6,
+ uniref50_transform_csv_7,
+ uniref50_transform_csv_8,
+ ]
  >> load_uniref50_to_bq_1
- >> uniref50_transform_csv_2
  >> load_uniref50_to_bq_2
- >> uniref50_transform_csv_3
  >> load_uniref50_to_bq_3
- >> uniref50_transform_csv_4
  >> load_uniref50_to_bq_4
- >> uniref50_transform_csv_5
  >> load_uniref50_to_bq_5
- >> uniref50_transform_csv_6
  >> load_uniref50_to_bq_6
- >> uniref50_transform_csv_7
  >> load_uniref50_to_bq_7
- >> uniref50_transform_csv_8
  >> load_uniref50_to_bq_8
  )