diff --git a/teste_dag.py b/teste_dag.py index 04b0585ba49163e46cd0f3da8a9eef1eceb6ead0..acecadea3be2efe661b620f0416a6b7b989c1e16 100644 --- a/teste_dag.py +++ b/teste_dag.py @@ -19,32 +19,29 @@ 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( - task_id="landing_to_bronze", - application=f"{repo_home}/gen_bronze.py", - application_args=[ - params["input_file"] - ], - packages="org.apache.hadoop:hadoop-aws:3.3.4" - ) + file = dag1.params["input_file"] - @task - def 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_spark = SparkSubmitOperator( + task_id="landing_to_bronze", + application=f"{repo_home}/gen_bronze.py", + application_args=[ + file + ], + packages="org.apache.hadoop:hadoop-aws:3.3.4" + ) - @task - def task_bronze_to_gold(): - BashOperator( - task_id="bronze_to_silver_to_gold", - cwd="/tmp/proj_teste", - bash_command="dbt deps && dbt build", - ) + 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_bronze_to_gold = BashOperator( + task_id="bronze_to_silver_to_gold", + cwd="/tmp/proj_teste", + bash_command="dbt deps && dbt build", + ) task_spark >> task_dbt_setup >> task_bronze_to_gold