Airflow 2.1.0 passing variable to another DAG using TriggerDagRunOperator

user_15739766 0
Viewed 1463

We're using Airflow 2.1.0 and want to trigger a DAG and pass a variable to it (an S3 file name) using TriggerDagRunOperator.

I've found examples of this and can pass a static JSON to the next DAG using conf:

    @task()
    def trigger_target_dag_task(context):
        TriggerDagRunOperator(
            task_id="trigger_target_dag",
            trigger_dag_id="target_dag",
            conf={"file_name": "test.txt"}
        ).execute(context)

However, I cannot find current examples where the conf is dynamically created without using python_callable - this seems close:

Airflow 2.0.0+ - Pass a Dynamically Generated Dictionary to DAG Triggered by TriggerDagRunOperator

https://github.com/apache/airflow/pull/6317#issuecomment-859556243

Is this possible?

Updated question:

This method did not work when I used:

    @task()
    def trigger_dag_task(context):
        TriggerDagRunOperator(
            task_id="trigger_dag_task",
            trigger_dag_id="target_dag",
            conf={"payload": "{{ ti.xcom_pull(task_ids='extract_rss') }}"},
        ).execute(context)

The target_dag received the conf as a string:

{logging_mixin.py:104} INFO - Remotely received value of {{ ti.xcom_pull(task_ids='extract_rss') }}
airflow
1 Answers

Conf is a templated field, so you could use Jinja to pass in any variable. Consider this example based on the official TriggerDagRunOperator example

If the variable (object_name) is within your scope you could do:

Controller DAG:

dag = DAG(
    dag_id="example_trigger_controller_dag",
    default_args={"owner": "airflow"},
    start_date=days_ago(2),
    schedule_interval="@once",
    tags=['example'],
)
object_name = "my-object-s3-aws"

trigger = TriggerDagRunOperator(
    task_id="test_trigger_dagrun",
    trigger_dag_id="example_trigger_target_dag",
    conf={"s3_object":  object_name},
    dag=dag,
)

Target DAG:

dag = DAG(
    dag_id="example_trigger_target_dag",
    default_args={"owner": "airflow"},
    start_date=days_ago(2),
    schedule_interval=None,
    tags=['example'],
)


def run_this_func(**context):
    print("Remotely received value of {} for key=message".format(
        context["dag_run"].conf["s3_object"]))


run_this = PythonOperator(
    task_id="run_this", python_callable=run_this_func, dag=dag)

bash_task = BashOperator(
    task_id="bash_task",
    bash_command='echo "Here is the message: $message"',
    env={'message': '{{ dag_run.conf["s3_object"] if dag_run else "" }}'},
    dag=dag,
)

If the variable is stored as an Airflow Variable you could retrieve it like this:

conf={"s3_object": "{{var.json.s3_object}}"}

If it were an XCom from a previous task, you could do:

conf={"s3_object": "{{ ti.xcom_pull(task_ids='previous_task_id', key='return_value') }}"

Let me know if that worked for you!

docs

Edit:

This is a working example, tested in version 2.0.1, using xcom_pull in conf param:

Controller DAG:

from airflow import DAG
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def _do_something():
    return "my-object-s3-aws"

dag = DAG(
    dag_id="example_trigger_controller_dag",
    default_args={"owner": "airflow"},
    start_date=days_ago(2),
    schedule_interval="@once",
    tags=['example'],
)

task_1 = PythonOperator(task_id='previous_task_id',
                        python_callable=_do_something)

trigger = TriggerDagRunOperator(
    task_id="test_trigger_dagrun",
    trigger_dag_id="example_trigger_target_dag",
    conf={
        "s3_object":
        "{{ ti.xcom_pull(task_ids='previous_task_id', key='return_value') }}"},
    dag=dag,
)

task_1 >> trigger

Target DAG:


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

dag = DAG(
    dag_id="example_trigger_target_dag",
    default_args={"owner": "airflow"},
    start_date=days_ago(2),
    schedule_interval=None,
    tags=['example'],
)


def run_this_func(**context):
    print("Remotely received value of {} ".format(
        context["dag_run"].conf["s3_object"]))


run_this = PythonOperator(
    task_id="run_this", python_callable=run_this_func, dag=dag)

bash_task = BashOperator(
    task_id="bash_task",
    bash_command='echo "Here is the message: $s3_object"',
    env={'s3_object': '{{ dag_run.conf["s3_object"] if dag_run else "" }}'},
    dag=dag,
)

Logs from run_this task:

[2021-07-15 19:24:11,410] {logging_mixin.py:104} INFO - Remotely received value of my-object-s3-aws

user_105692200
Related