Airflow中创建与父DAG不同调度间隔的SubDAG问题咨询

嘿，这个问题我太熟悉了！先给你明确答案：这确实是Airflow的默认行为。

SubDAG在Airflow里的定位其实是「父DAG内的任务分组容器」，它并不是一个独立的调度单元——也就是说，你给SubDAG设置的schedule_interval参数完全不会被Airflow的调度器读取。只要父DAG被触发运行，SubDAG里的任务就会按照父DAG内部的依赖关系启动，完全无视自己的调度配置。

接下来是你要的「不用转独立DAG也不用传感器」的解决方法：

方法1：利用父DAG的任务依赖+PythonOperator做时间判断 #

你可以在每个SubDAG的入口处，加一个PythonOperator任务，用来判断当前的执行时间是否符合SubDAG的预期运行时间，如果不符合就直接标记为成功（跳过后续任务），符合的话再执行SubDAG的实际逻辑。

举个简单的例子：

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.subdag import SubDagOperator
from subdag_1 import build_subdag_1
from subdag_2 import build_subdag_2

def check_run_time_1(**context):
    # 定义SubDAG1的预期运行时间：每天凌晨1点
    target_hour = 1
    current_execution_hour = context['execution_date'].hour
    if current_execution_hour != target_hour:
        # 如果不是目标时间，直接跳过后续任务
        return "Skip: Not target hour"
    else:
        return "Proceed: Target hour reached"

def check_run_time_2(**context):
    # 定义SubDAG2的预期运行时间：每天凌晨3点
    target_hour = 3
    current_execution_hour = context['execution_date'].hour
    if current_execution_hour != target_hour:
        return "Skip: Not target hour"
    else:
        return "Proceed: Target hour reached"

with DAG(
    dag_id='parent_dag',
    schedule_interval='@hourly',  # 父DAG每小时调度一次，覆盖目标时间点
    start_date=datetime(2024,1,1),
    catchup=False
) as dag:

    check_time_1 = PythonOperator(
        task_id='check_time_for_subdag1',
        python_callable=check_run_time_1,
        provide_context=True
    )

    check_time_2 = PythonOperator(
        task_id='check_time_for_subdag2',
        python_callable=check_run_time_2,
        provide_context=True
    )

    subdag_1 = SubDagOperator(
        task_id='subdag_1',
        subdag=build_subdag_1('parent_dag', 'subdag_1', dag.start_date, '@daily'),
        # 注意：这里SubDAG的schedule_interval随便写，不会生效
    )

    subdag_2 = SubDagOperator(
        task_id='subdag_2',
        subdag=build_subdag_2('parent_dag', 'subdag_2', dag.start_date, '@daily'),
    )

    # 设置依赖：检查时间通过后才执行SubDAG
    check_time_1 >> subdag_1
    check_time_2 >> subdag_2

这里要注意：父DAG的调度频率必须覆盖SubDAG的运行时间点，比如示例中用@hourly，确保到了1点、3点的那次父DAG执行能触发检查逻辑。

方法2：Airflow 2.x+ 用TaskFlow API分支任务控制 #

如果你用的是Airflow 2.x及以上版本，用TaskFlow API可以更简洁地实现逻辑，用分支任务决定是否执行对应的任务组（替代传统SubDAG）：

from datetime import datetime
from airflow.decorators import dag, task, task_group

@task
def should_run_subdag_1(execution_date: datetime):
    # 判断是否是SubDAG1的目标运行小时
    return execution_date.hour == 1

@task
def should_run_subdag_2(execution_date: datetime):
    # 判断是否是SubDAG2的目标运行小时
    return execution_date.hour == 3

@task_group(group_id='subdag_1_group')
def subdag_1_tasks():
    # 这里写SubDAG1的所有任务逻辑
    @task
    def task1():
        print("Running SubDAG1 task 1")
    @task
    def task2():
        print("Running SubDAG1 task 2")
    task1() >> task2()

@task_group(group_id='subdag_2_group')
def subdag_2_tasks():
    # 这里写SubDAG2的所有任务逻辑
    @task
    def task_a():
        print("Running SubDAG2 task A")
    @task
    def task_b():
        print("Running SubDAG2 task B")
    task_a() >> task_b()

@dag(schedule_interval='@hourly', start_date=datetime(2024,1,1), catchup=False)
def parent_dag():
    run_sub1 = should_run_subdag_1()
    run_sub2 = should_run_subdag_2()

    # 只有当分支返回True时，才执行对应的任务组
    run_sub1 >> subdag_1_tasks()
    run_sub2 >> subdag_2_tasks()

parent_dag()

这种方式用task_group替代了传统SubDagOperator，逻辑更直观，不需要额外依赖传感器，完全通过Python判断执行时间来控制任务组的启动。

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