如何在Airflow单个Dag中配置不同任务执行间隔及依赖？

嘿，这个场景我之前在项目里处理过好几次，咱们一步步来理清思路和解决方案：

首先得明确一个关键点：直接修改单个任务的schedule_interval是不可行的——Airflow里任务的调度周期默认是继承所属DAG的，单独给任务设置这个参数在绝大多数版本里都不会生效（Airflow 2.2+虽有任务级调度的实验性支持，但配置复杂，完全没必要用在这种简单场景里）。

针对你的需求，这里有两种成熟的最佳实践方案，你可以根据自己的偏好选择：

方案一：单DAG内通过分支判断控制任务D的执行 #

把整个DAG的schedule_interval设为@daily（每日运行），然后在任务A之后加一个判断任务，只有当当天是你指定的每周日期（比如周一）时，才触发任务D执行，否则直接跳过。

传统Operator写法示例 #

from airflow import DAG
from airflow.operators.python import ShortCircuitOperator, PythonOperator
from datetime import datetime, timedelta
import pendulum

def should_run_weekly(**context):
    # 这里判断当前执行日期是否为每周一，你可以改成自己需要的日期（0=周一，6=周日）
    execution_date = context['execution_date']
    return execution_date.weekday() == 0

default_args = {
    'owner': 'airflow',
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'daily_with_weekly_task',
    default_args=default_args,
    description='Daily tasks with weekly dependent task',
    schedule_interval='@daily',
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
) as dag:

    task_a = PythonOperator(
        task_id='task_a',
        python_callable=lambda: print("Running Task A")
    )

    task_b = PythonOperator(
        task_id='task_b',
        python_callable=lambda: print("Running Task B")
    )

    task_c = PythonOperator(
        task_id='task_c',
        python_callable=lambda: print("Running Task C")
    )

    check_weekly_run = ShortCircuitOperator(
        task_id='check_weekly_run',
        python_callable=should_run_weekly,
        provide_context=True
    )

    task_d = PythonOperator(
        task_id='task_d',
        python_callable=lambda: print("Running Task D")
    )

    # 设置依赖：A执行完后判断是否要跑D，A、B、C并行执行
    task_a >> check_weekly_run >> task_d
    [task_a, task_b, task_c]

Airflow 2.x TaskFlow API写法示例 #

如果用的是Airflow 2.x的TaskFlow，代码会更简洁：

from airflow.decorators import dag, task
from datetime import datetime, timedelta
import pendulum

default_args = {
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

@dag(
    schedule_interval='@daily',
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    default_args=default_args,
    description='Daily tasks with weekly dependent task (TaskFlow)'
)
def daily_weekly_dag():

    @task
    def task_a():
        print("Running Task A")
        return "A done"

    @task
    def task_b():
        print("Running Task B")

    @task
    def task_c():
        print("Running Task C")

    @task
    def should_run_weekly(execution_date):
        # 同样判断是否为周一
        return execution_date.weekday() == 0

    @task
    def task_d():
        print("Running Task D")

    # 任务依赖逻辑
    a_result = task_a()
    run_d_flag = should_run_weekly(a_result.execution_date)
    # 只有当run_d_flag返回True时，task_d才会执行
    run_d_flag >> task_d()
    # A、B、C并行启动
    [a_result, task_b(), task_c()]

dag = daily_weekly_dag()

方案二：拆分两个独立DAG（我最推荐的最佳实践） #

把每日任务和每周任务拆成两个独立的DAG，用ExternalTaskSensor让每周的任务D等待每日DAG里的任务A执行成功后再启动。这种方案更符合Airflow的设计理念，维护和监控起来都更方便。

每日任务DAG（daily_tasks_dag.py） #

这个DAG只负责跑A、B、C，每日执行：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import pendulum

default_args = {
    'owner': 'airflow',
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'daily_tasks',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
) as dag:

    task_a = PythonOperator(
        task_id='task_a',
        python_callable=lambda: print("Running Task A")
    )

    task_b = PythonOperator(
        task_id='task_b',
        python_callable=lambda: print("Running Task B")
    )

    task_c = PythonOperator(
        task_id='task_c',
        python_callable=lambda: print("Running Task C")
    )

    # A、B、C无依赖并行执行
    [task_a, task_b, task_c]

每周任务DAG（weekly_task_d.py） #

这个DAG只负责跑D，每周执行，并且会先等待每日DAG的A完成：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime, timedelta
import pendulum

default_args = {
    'owner': 'airflow',
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'weekly_task_d',
    default_args=default_args,
    schedule_interval='@weekly',  # 每周执行一次
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
) as dag:

    # 等待daily_tasks DAG中的task_a成功完成
    wait_for_task_a = ExternalTaskSensor(
        task_id='wait_for_task_a',
        external_dag_id='daily_tasks',
        external_task_id='task_a',
        execution_delta=timedelta(days=0),  # 等待同一天的daily_tasks执行完成
        mode='reschedule',
        timeout=3600  # 超时时间，可根据实际调整
    )

    task_d = PythonOperator(
        task_id='task_d',
        python_callable=lambda: print("Running Task D")
    )

    wait_for_task_a >> task_d

方案对比与选择建议 #

• 单DAG分支方案：适合逻辑简单、希望所有任务都在同一个DAG里管理的场景，但如果后续任务增多，DAG会变得臃肿，不好维护。
• 双DAG拆分方案：强烈推荐，它把每日和每周的任务逻辑完全分离，你可以单独调整每个DAG的调度周期、监控状态，甚至给不同DAG设置不同的权限，符合Airflow的模块化设计原则，也降低了单个DAG故障的影响范围。

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