如何基于DAG参数动态生成任务并触发子DAG？

问题背景 #

需要通过DAG参数传入日期字符串列表，为每个日期生成任务并按顺序触发子DAG。原代码尝试用Jinja模板遍历参数列表失败，修改为TaskFlow动态映射后，子DAG仍未触发。

原代码问题分析 #

原代码中直接在循环里使用{{params.date_list}}，这是Jinja模板语法，仅在任务运行时渲染，但DAG解析阶段（静态初始化）会把它当作普通字符串处理，循环遍历的是字符串"{{params.date_list}}"的每个字符，而非实际的日期列表，导致生成的任务不符合预期。

原代码片段：

with DAG(
       ...,
       params={"date_list": Param(["2022-10-10"], type="array", ...)},
) as dag:
    t0 = None
    for idx, _date in enumerate("{{params.date_list}}"):
        t1 = DummyOperator(task_id=f"{idx}-{_date}")
       
        if t0 is not None:
            t0 >> t1
        t0 = t1

修改后代码的核心问题 #

修改后的代码存在两个关键错误：

1. TaskFlow任务返回Operator实例无效：@task装饰的函数返回TriggerDagRunOperator实例没有意义，TaskFlow任务的返回值仅用于XCom传递，Airflow不会自动执行该Operator，必须在任务内部手动调用execute方法触发子DAG。
2. make_list函数调用时机错误：直接在DAG解析阶段调用get_current_context()无法获取运行时的context（包括params），必须将make_list包装为@task，在任务运行时才能正确获取参数。

修改后的错误代码片段：

def make_list():
    context = get_current_context()
    return context["params"]["date_list"]

@task
def generate_tasks(arg):
    return TriggerDagRunOperator(task_id=f"{arg}", trigger_dag_id="test_action", wait_for_completion=True)

generate_tasks = generate_tasks.expand(arg=make_list())
(generate_tasks)

正确解决方案 #

方案1：TaskFlow动态映射+手动触发子DAG #

将参数获取和子DAG触发都包装为TaskFlow任务，在任务内部手动执行TriggerDagRunOperator的execute方法，同时实现任务的顺序依赖：

from airflow.decorators import dag, task
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.utils.context import get_current_context
from datetime import datetime
from airflow.models.param import Param

@dag(
    start_date=datetime(2023, 1, 1),
    params={"date_list": Param(["2022-01-10", "2023-06-22"], type="array")},
    schedule=None,
    catchup=False
)
def main_dag():
    # 运行时获取日期列表参数
    @task
    def get_date_list():
        context = get_current_context()
        return context["params"]["date_list"]

    # 触发子DAG的任务
    @task
    def trigger_subdag(date):
        context = get_current_context()
        # 创建TriggerDagRunOperator并手动执行
        trigger_op = TriggerDagRunOperator(
            task_id=f"trigger_test_action_{date}",
            trigger_dag_id="test_action",
            conf={"target_date": date},  # 传递日期参数给子DAG
            wait_for_completion=True,
            poke_interval=60
        )
        trigger_op.execute(context=context)

    # 动态生成任务
    date_list = get_date_list()
    trigger_tasks = trigger_subdag.expand(date=date_list)

    # 设置顺序执行：每个任务依赖前一个任务完成
    for i in range(1, len(trigger_tasks)):
        trigger_tasks[i-1] >> trigger_tasks[i]

main_dag()

方案2：DAG解析阶段动态生成任务（仅适用于静态参数） #

如果日期列表是固定的静态参数，可直接在DAG解析阶段循环生成TriggerDagRunOperator并设置依赖：

from airflow import DAG
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from datetime import datetime
from airflow.models.param import Param

with DAG(
    dag_id="main_dag_static",
    start_date=datetime(2023, 1, 1),
    params={"date_list": Param(["2022-01-10", "2023-06-22"], type="array")},
    schedule=None,
    catchup=False
) as dag:
    # 注意：这里的params是静态默认值，若运行时传入新参数，此方法无法获取
    date_list = dag.params["date_list"]
    prev_task = None
    for idx, date in enumerate(date_list):
        trigger_task = TriggerDagRunOperator(
            task_id=f"{idx}-{date}",
            trigger_dag_id="test_action",
            conf={"target_date": date},
            wait_for_completion=True
        )
        if prev_task:
            prev_task >> trigger_task
        prev_task = trigger_task

关键说明 #

• 若需要运行时动态传入参数，必须使用方案1的TaskFlow动态映射，因为DAG解析阶段无法获取运行时的params。
• 子DAGtest_action需要确保已正确部署，且trigger_dag_id与子DAG的dag_id完全一致。
• wait_for_completion=True会让主任务等待子DAG执行完成后再继续，适合顺序执行的场景。

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