Airflow中跨任务传递dag_run_id失败，bash_task无法打印该值

你当前用全局变量dag_run_id的方式无效，因为Airflow任务可能运行在不同的Worker进程中，全局变量无法跨进程共享，导致bash_task无法获取到该值。以下是两种可行的解决方案：

方案1：直接使用Airflow内置模板变量 #

Airflow原生提供了dag_run模板上下文变量，无需额外编写Python任务获取，直接在BashOperator中引用即可：

bash_task = BashOperator(
    task_id='bash_task',
    bash_command='echo {{ dag_run.run_id }}',
    dag=dag
)

这样bash_task就能直接打印当前DAG Run的run_id。

方案2：通过XCom传递（适合需要自定义处理run_id的场景） #

如果需要在Python任务中对run_id做额外处理后再传递给其他任务，可使用Airflow的XCom机制实现跨任务数据共享：

步骤1：在Python任务中推送XCom #

def my_function(**kwargs):
    dag_run_id = kwargs['dag_run'].run_id
    # 自定义key推送XCom，方便后续任务精准获取
    kwargs['ti'].xcom_push(key='target_dag_run_id', value=dag_run_id)

example_task = PythonOperator(
    task_id='example_task',
    python_callable=my_function,
    provide_context=True,
    dag=dag
)

步骤2：在Bash任务中拉取XCom #

在BashOperator中通过模板语法获取指定任务推送的XCom值，同时要确保任务执行顺序：

bash_task = BashOperator(
    task_id='bash_task',
    bash_command='echo {{ ti.xcom_pull(task_ids="example_task", key="target_dag_run_id") }}',
    dag=dag
)

# 设置任务依赖，保证example_task先执行
example_task >> bash_task

内容的提问来源于stack exchange，提问作者Jangyasmita Priyadarshini