如何从Airflow任务创建DAG？动态生成带参数子DAG需求咨询

我来给你梳理下怎么实现这个需求——毕竟Airflow的DAG是调度器解析文件时创建的，而父DAG任务的参数是运行时才生成的，得用「存储参数+动态生成DAG脚本」的组合方式来搞定，具体步骤如下：

1. 先搞定父DAG：生成参数并持久化存储 #

首先，父DAG里的任务得把生成的params1、params2、params3存起来，我推荐用Airflow的Variable（全局持久化，跨DAG共享方便），用PythonOperator来实现参数生成和存储：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime

def generate_and_save_params():
    # 这里模拟生成非固定参数，实际场景你可以换成从数据库/API拉取的逻辑
    generated_params = {
        "params1": "your_value_1",
        "params2": "your_value_2",
        "params3": "your_value_3"
    }
    # 把参数序列化成JSON存到Variable里，key统一用dynamic_dag_params
    Variable.set("dynamic_dag_params", generated_params, serialize_json=True)
    
    # 重要：触发Airflow重新解析DAG文件，这样新的子DAG才能被调度器识别
    # 这里用subprocess调用Airflow CLI命令，你也可以换成Airflow API调用
    import subprocess
    subprocess.run(["airflow", "dags", "reserialize"], check=True)

with DAG(
    dag_id="parent_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as parent_dag:
    gen_params_task = PythonOperator(
        task_id="generate_and_save_params",
        python_callable=generate_and_save_params
    )

2. 写动态生成子DAG的脚本 #

接下来单独写一个DAG文件（比如叫dynamic_child_dags.py），这个文件会在Airflow调度器解析时，读取刚才存在Variable里的参数，自动生成对应的子DAG：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime

# 读取存储的参数，加个异常处理防止首次运行时Variable不存在
try:
    dynamic_params = Variable.get("dynamic_dag_params", deserialize_json=True)
except Variable.DoesNotExist:
    dynamic_params = {}

# 遍历每个参数，为每个参数创建一个独立的子DAG
for param_key, param_value in dynamic_params.items():
    # 子DAG里的任务函数，能直接拿到对应参数
    def execute_child_task(**context):
        print(f"当前子DAG使用的参数：{param_key} = {param_value}")
        # 这里写你的业务逻辑，比如调用BigQuery任务、处理数据等等
        # 也可以通过context获取Airflow的上下文信息，比如执行时间、任务ID之类的

    # 创建子DAG
    with DAG(
        dag_id=f"child_dag_{param_key}",
        start_date=datetime(2024, 1, 1),
        schedule_interval="@daily",
        catchup=False,
        # 把参数放到default_args里，方便任务随时获取
        default_args={"params": {param_key: param_value}}
    ) as child_dag:
        run_child_task = PythonOperator(
            task_id="run_child_business_logic",
            python_callable=execute_child_task,
            provide_context=True
        )

3. 几个关键注意点要记牢 #

• 参数存储选Variable还是XCom？：我推荐Variable，因为XCom是和父DAG的某次运行实例绑定的，而Variable是全局持久化的，适合动态DAG读取；如果你的子DAG只需要和父DAG的某次运行绑定，那可以用XCom+TriggerDagRunOperator的方式（后面给你补充这种方案）。
• 一定要触发DAG重新解析：父DAG生成参数后，必须让Airflow重新解析DAG文件，不然调度器看不到新生成的子DAG，用airflow dags reserialize或者API都可以。
• 子DAG ID要唯一：这里用param_key作为子DAG ID的后缀，保证每个子DAG都有唯一标识，避免冲突。
• 参数更新的处理：如果父DAG重新运行生成了新参数，动态脚本下次解析时会更新子DAG，旧的子DAG会被移除；如果需要保留历史子DAG，你可以把参数存储改成列表结构，记录每次生成的参数。

补充：如果参数数量固定，用TriggerDagRunOperator更简单 #

要是你确定每次都是生成3个参数，不需要动态创建新的DAG文件，那可以预先创建3个子DAG，然后父DAG运行时用TriggerDagRunOperator触发它们，把参数通过conf传递过去：

父DAG里的触发逻辑 #

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

def generate_params(**context):
    params = {"params1": "val1", "params2": "val2", "params3": "val3"}
    # 把参数推送到XCom，供子DAG读取
    context["ti"].xcom_push(key="dynamic_params", value=params)

with DAG("parent_dag", ...) as parent_dag:
    gen_params = PythonOperator(
        task_id="generate_params",
        python_callable=generate_params,
        provide_context=True
    )
    # 逐个触发子DAG，传递参数标识
    for param_name in ["params1", "params2", "params3"]:
        trigger_child = TriggerDagRunOperator(
            task_id=f"trigger_child_{param_name}",
            trigger_dag_id=f"child_dag_{param_name}",
            conf={"target_param": param_name},
            provide_context=True
        )
        gen_params >> trigger_child

子DAG读取参数的逻辑 #

def run_child_task(**context):
    # 从conf里拿到要获取的参数名
    param_name = context["dag_run"].conf.get("target_param")
    # 从父DAG的XCom里拉取对应参数值
    param_value = context["ti"].xcom_pull(
        dag_id="parent_dag", 
        task_ids="generate_params", 
        key="dynamic_params"
    )[param_name]
    print(f"拿到的参数：{param_name} = {param_value}")
    # 写你的业务逻辑

with DAG("child_dag_params1", ...) as child_dag:
    PythonOperator(
        task_id="run_child_task",
        python_callable=run_child_task,
        provide_context=True
    )

这种方式不需要动态生成DAG文件，适合参数数量固定但值不固定的场景，维护起来更简单。

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