Can I use a TriggerDagRunOperator to pass a parameter to the triggered dag? Airflow

user_4991324 0
Viewed 7968

Im using Airflow 1.10.11

I can read that I can send parameters to a dag execution using the UI and CLI ( https://airflow.apache.org/docs/apache-airflow/1.10.11/dag-run.html ), but can I do it using a TriggerDagRunOperator ? ( https://airflow.apache.org/docs/apache-airflow/1.10.11/_api/airflow/operators/dagrun_operator/index.html ).

Reading https://github.com/apache/airflow/blob/master/airflow/example_dags/example_trigger_target_dag.py , https://github.com/apache/airflow/blob/master/airflow/example_dags/example_trigger_controller_dag.py it seems like I can, but when I tried this script myself I was not able to retrieve the conf message (probably because it seems they are using airflow 2.0 while I am using 1.10).

So my question is: is there a way to send parameter to the target dag when TriggerDagRunOperator that target?

Thanks in advance

airflow python
2 Answers

Here is an example that demonstrates how to set the conf sent with dagruns triggered by TriggerDagRunOperator (in 1.10.11).

trigger.py

from datetime import datetime
from airflow.models import DAG
from airflow.operators.dagrun_operator import TriggerDagRunOperator

dag = DAG(
    dag_id='trigger',
    schedule_interval='@once',
    start_date=datetime(2021, 1, 1)
)


def modify_dro(context, dagrun_order):
    print(context)
    print(dagrun_order)
    dagrun_order.payload = {
        "message": "This is my conf message"
    }
    return dagrun_order


run_this = TriggerDagRunOperator(
    task_id='run_this',
    trigger_dag_id='my_dag',
    python_callable=modify_dro,
    dag=dag
)

dag.py

from datetime import datetime
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator

dag = DAG(
    dag_id='my_dag',
    schedule_interval='@once',
    start_date=datetime(2021, 1, 1)
)


def run_this_func(**context):
    print(context["dag_run"].conf)


run_this = PythonOperator(
    task_id='run_this',
    provide_context=True,
    python_callable=run_this_func,
    dag=dag,
)

Here is the output from the run_this task in the my_dag DAG

[2021-03-05 16:39:15,649] {logging_mixin.py:112} INFO - {'message': 'This is my conf message'}

The TriggerDagRunOperator takes in a python_callable that allows you to modify the DagRunOrder. DagRunOrder is an abstract construct that holds the run id and payload which will be sent as conf when the target DAG is triggered.

user_14649660
user_31526540
Related