WIP

Ferlab-Ste-Justine · Mar 8, 2024 · 3807e32 · 3807e32
1 parent 446192f
commit 3807e32
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 11 deletions.
diff --git a/dags/etl.py b/dags/etl.py
@@ -47,4 +47,4 @@ def publish_operator(name:str):
                         )            
 
        publish_operator('study') >> publish_operator('participant') >> publish_operator('file') >>  publish_operator('biospecimen') 
-    fhavro_export() >> etl_import >> prepare_index >> es_templates_update >> index >> publish >> arranger_task
+    fhavro_export() >> etl_import() >> prepare_index >> es_templates_update >> index >> publish >> arranger_task
diff --git a/dags/etl_import.py b/dags/etl_import.py
@@ -4,15 +4,16 @@
 from lib.config import etl_base_config, spark_small_conf, import_jar, default_config_file, release_id, study_ids
 from lib.operators.spark import SparkOperator
 
-etl_import = etl_base_config \
-    .with_spark_class('bio.ferlab.fhir.etl.ImportTask') \
-    .with_spark_jar(import_jar) \
-    .add_spark_conf(spark_small_conf) \
-    .args(default_config_file, 'default', release_id, study_ids) \
-    .operator(
-            task_id='import_task',
-            name='etl-import-task'
-        )    
+def etl_import():
+    return etl_base_config \
+        .with_spark_class('bio.ferlab.fhir.etl.ImportTask') \
+        .with_spark_jar(import_jar) \
+         .add_spark_conf(spark_small_conf) \
+         .args(default_config_file, 'default', release_id, study_ids) \
+         .operator(
+              task_id='import_task',
+              name='etl-import-task'
+          )    
 
 with DAG(
         dag_id='import',
@@ -24,4 +25,4 @@
             'project': Param('cqdg', type='string'),
         },
 ) as dag:
-    etl_import
+    etl_import()