Airflow如何在单次DAG运行中对多个动态日期执行相同查询

不需要动态生成子DAG，也不需要在算子内部构造新的DAG/算子，Airflow原生能力完全可以覆盖这类需求，以下是经过生产验证的可行方案：

方案1：直接用Jinja模板传参（最推荐，无额外依赖） #

首先明确一个容易踩的认知误区：所有Operator标记为模板渲染的字段，默认就可以直接访问execution_date、macros等上下文变量，完全不需要额外配置user_defined_macros。
BigQueryOperator（新版对应BigQueryInsertJobOperator）的sql、目标表配置、查询参数配置本身就是模板字段，你需要的日期计算、分区写入逻辑直接写在模板表达式里即可：

• 如果要跑的日期是基于execution_date的固定规则（比如执行日前3天、近7天的周末等），直接在DAG顶层循环生成对应数量的BQ任务即可，循环时直接嵌入Jinja日期计算表达式：

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="bq_multi_date_demo",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False
) as dag:
    # 枚举需要跑的日期偏移量，比如-1、-2、-3对应执行日前3天
    for offset in [-1, -2, -3]:
        BigQueryInsertJobOperator(
            task_id=f"bq_run_offset_{offset}",
            configuration={
                "query": {
                    # 直接引用外部SQL文件
                    "query": "{% include 'your_external_query.sql' %}",
                    "useLegacySql": False,
                    # 把计算好的日期作为命名参数传入SQL，不需要自定义宏
                    "queryParameters": [
                        {
                            "name": "process_date",
                            "parameterType": {"type": "DATE"},
                            "parameterValue": {
                                "value": "{{ (execution_date + macros.timedelta(days=offset)).strftime('%Y-%m-%d') }}"
                            }
                        }
                    ]
                },
                # 直接拼接分区后缀，模板自动渲染为对应日期分区
                "destinationTable": {
                    "projectId": "your_project",
                    "datasetId": "your_dataset",
                    "tableId": f"target_table${{{{ (execution_date + macros.timedelta(days={offset})).strftime('%Y%m%d') }}}}"
                }
            }
        )

外部SQL文件里直接用@process_date就能拿到传入的日期值，不需要额外在DAG层做任何宏配置。如果日期计算逻辑复杂，直接在Jinja表达式里调用macros下的日期方法即可，覆盖绝大多数计算场景。

方案2：动态任务映射适配运行时才确定的日期列表 #

如果需要跑的日期列表没有固定偏移规则，必须运行时结合execution_date动态计算，直接用Airflow 2.3+版本自带的动态任务映射能力即可，不需要在算子内部生成子DAG：

1. 先用一个Python任务基于execution_date计算得到所有需要处理的日期列表，作为返回值
2. 将日期列表映射展开为多个独立的BQ任务实例，每个实例自动拿到对应日期参数，写入对应分区
   参考代码片段：

from airflow.decorators import task
from datetime import timedelta

with DAG(
    dag_id="bq_dynamic_mapping_demo",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False
) as dag:
    @task
    def calc_target_dates(execution_date=None):
        # 算子内部可以写任意复杂的日期筛选逻辑，返回需要处理的日期列表
        res = []
        for i in range(1, 14):
            current = execution_date.date() + timedelta(days=-i)
            # 示例：筛选近14天内的周一、周五
            if current.weekday() in [0, 4]:
                res.append(current.strftime("%Y-%m-%d"))
        return res

    dates = calc_target_dates()

    # 为每个日期自动生成独立的BQ任务实例
    BigQueryInsertJobOperator.partial(
        task_id="run_bq_per_date",
        configuration={
            "query": {
                "query": "{% include 'your_external_query.sql' %}",
                "useLegacySql": False
            }
        }
    ).expand(
        op_kwargs=dates.map(lambda dt: {
            "configuration.query.queryParameters": [
                {"name": "process_date", "parameterType": {"type": "DATE"}, "parameterValue": {"value": dt}}
            ],
            "configuration.destinationTable.tableId": f"target_table${dt.replace('-', '')}"
        })
    )

额外说明 #

旧版本SubDAG因为存在死锁、上下文隔离等问题已经被官方废弃，如果需要做任务分组直接用TaskGroup即可，TaskGroup内的任务和顶层任务一样可以直接访问所有上下文变量，不需要单独配置user_defined_macros。

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