teste

teste_dag.py

@@ -19,28 +19,25 @@ with DAG(
         "input_file": Param("invalid_file.jsonl", type="string"),
     },
 ) as dag1:
+    dag1.log.info(f"Input file {dag1.params["input_file"]}")
     
-    @task
-    def task_spark(params: dict):
-        SparkSubmitOperator(
+    file = dag1.params["input_file"]
+
+    task_spark = SparkSubmitOperator(
         task_id="landing_to_bronze",
         application=f"{repo_home}/gen_bronze.py",
         application_args=[
-                params["input_file"]
+            file
         ],
         packages="org.apache.hadoop:hadoop-aws:3.3.4"
     )
 
-    @task
-    def task_dbt_setup():
-        BashOperator(
+    task_dbt_setup = BashOperator(
         task_id="dbt_setup",
         bash_command=f"rm -rf /tmp/proj_teste && cp -r {repo_home}/proj_teste /tmp/proj_teste",
     )
 
-    @task
-    def task_bronze_to_gold():
-        BashOperator(
+    task_bronze_to_gold = BashOperator(
         task_id="bronze_to_silver_to_gold",
         cwd="/tmp/proj_teste",
         bash_command="dbt deps && dbt build",