如何通过Airflow TaskFlow API在Dataset关联的跨DAG任务间传递数据？

你当前遇到的问题是直接引用first_dag.tasks[0].output会得到未解析的模板字符串，因为跨DAG的任务输出无法在DAG解析阶段直接获取实际运行值。以下是两种可行的解决方案：

方案一：通过XCom跨DAG拉取数据 #

在任务运行时，利用Airflow自动注入的TaskInstance对象拉取第一个DAG的XCom数据：

修改第二个DAG的代码：

@dag(
    dag_id="xcom_test_end",
    default_args=default_dag_arguments,
    schedule=[ uiflow_xcom_test_dataset ],
    tags=["utility"]
)
def xcom_test_end():

    @task
    def pull_data(ti):
        # 拉取第一个DAG中push_data任务的返回值（默认存在XCom的return_value键下）
        data = ti.xcom_pull(task_ids='push_data', dag_id='xcom_test_start', key='return_value')
        print(f"data pulled is {data}")

    pull_data()

xcom_test_end()

• 说明：ti是Airflow自动注入的TaskInstance对象，通过它可以精准拉取指定DAG和任务的XCom数据。
• 注意：确保第一个DAG的push_data任务已成功运行并写入XCom，且两个DAG处于同一Airflow环境。

方案二：利用Dataset事件传递数据（推荐） #

Dataset本身支持更新时携带自定义数据，这种方式更贴合Dataset触发的设计逻辑，无需依赖跨DAG的XCom查询，数据与触发事件强关联，可靠性更高：

修改第一个DAG的push_data任务： #

@dag(
    dag_id="xcom_test_start",
    default_args=default_dag_arguments,
    schedule_interval="@once",
    start_date=pendulum.datetime(2024, 1, 1, tz="US/Eastern"),
)
def xcom_test_start():

    @task
    def push_data():
        from airflow.datasets import Dataset
        # 更新Dataset时携带自定义数据
        ds = Dataset("uiflow_xcom_test")
        ds.update(data="sample data")
        return "sample data"

    push_data()

first_dag = xcom_test_start()

修改第二个DAG的pull_data任务： #

@dag(
    dag_id="xcom_test_end",
    default_args=default_dag_arguments,
    schedule=[ uiflow_xcom_test_dataset ],
    tags=["utility"]
)
def xcom_test_end():

    @task
    def pull_data(dataset_events):
        # 遍历触发当前DAG的Dataset事件，提取目标数据
        for event in dataset_events:
            if event.dataset.uri == "uiflow_xcom_test":
                data = event.data
                print(f"data pulled is {data}")
                break

    pull_data()

xcom_test_end()

• 说明：dataset_events是Airflow自动注入的参数，包含所有触发当前DAG运行的Dataset事件对象，可直接从中获取触发时携带的数据。
• 优势：无需硬编码目标DAG和任务ID，避免拉取到旧的XCom数据。

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