如何在Airflow 2+中通过动态参数触发Cloud Data Fusion并行任务

核心方案：使用Airflow动态任务映射（Dynamic Task Mapping） #

Airflow 2.3+引入的Dynamic Task Mapping完美适配你这种「基于上游任务输出动态生成并行任务」的场景，不需要在DAG解析阶段做循环，而是在运行时根据read_bq的输出自动生成并行的CloudDataFusionStartPipeline任务。

步骤1：确保read_bq任务正确输出XCom #

调整你的read_bq Python任务，让它返回包含所有流水线信息的列表，每个元素为字典，包含流水线名称和运行时参数：

from airflow.operators.python import PythonOperator

def read_bq():
    # 替换为你的BigQuery读取逻辑
    pipelines = [
        {"pipeline_name": "pipeline_1", "runtime_params": {"param1": "val1"}},
        {"pipeline_name": "pipeline_2", "runtime_params": {"param2": "val2"}},
        # 更多流水线配置...
    ]
    return pipelines

read_bq_task = PythonOperator(
    task_id="read_bq",
    python_callable=read_bq,
    dag=dag
)

任务返回值会自动推送到XCom，供后续任务直接引用。

步骤2：用expand方法动态生成并行任务 #

使用CloudDataFusionStartPipelineOperator的expand方法，基于read_bq_task的输出，为每个流水线生成独立的任务实例：

from airflow.providers.google.cloud.operators.datafusion import CloudDataFusionStartPipelineOperator

trigger_pipelines = CloudDataFusionStartPipelineOperator.partial(
    task_id="trigger_datafusion_pipeline",
    location="your-datafusion-location",
    project_id="your-gcp-project-id",
    instance_name="your-datafusion-instance-name"
).expand(
    pipeline_name=lambda x: x["pipeline_name"],
    runtime_parameters=lambda x: x["runtime_params"],
    trigger_rule="all_success",
    input=read_bq_task.output
)

# 设置任务依赖
read_bq_task >> trigger_pipelines

• partial：配置所有任务实例共享的固定参数（如地域、项目ID、DataFusion实例名）
• expand：
  • input指定上游任务的XCom输出作为数据源
  • 用lambda函数从每个列表元素中提取pipeline_name和runtime_parameters
  • 最终会生成与read_bq返回列表长度一致的并行任务

关键说明 #

• 无需手动XCom Pull：Dynamic Task Mapping自动处理上游输出传递，避免在DAG中写循环或手动拉取XCom
• 并行执行：所有动态生成的任务默认并行运行，可通过max_active_tis_per_dag等参数控制并发数
• 版本要求：确保Airflow版本≥2.3，google-cloud provider版本≥3.0.0

备选方案（Airflow <2.3） #

若你的Airflow版本低于2.3，无法使用动态映射，可采用以下方式：

1. 在read_bq任务中，将流水线信息写入临时存储（如GCS文件、Redis）
2. 新增一个PythonOperator中间任务，读取临时存储的流水线列表，动态生成CloudDataFusionStartPipelineOperator实例
3. 注意：这种方式需在运行时修改DAG结构，维护成本较高，优先推荐升级Airflow版本使用动态映射

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