Airflow如何设置跨不同execution_date的多任务依赖？

你的问题核心在于：同一个DAG里不能存在多个task_id相同的任务——哪怕你指定了不同的execution_date，Airflow依然会通过task_id识别任务唯一性，所以重复创建task_id='extraction'的DummyOperator才会触发“extraction already registered”的错误。

要实现你需要的跨日期依赖（复用昨日已完成的抽取、拟合任务），正确的做法是引用已有任务的历史Task Instance，而不是重新创建同名任务。下面是具体的实现步骤和代码示例：

1. 先定义基础任务（每个task_id只定义一次） #

首先我们把三个核心任务都定义为DAG里的基础任务，每个task_id只存在一个实例：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2018, 5, 17),  # 要早于第一个需要引用的昨日日期
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'model_training_pipeline',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

# 抽取任务（仅定义一次，对应当日的实例）
extraction = DummyOperator(
    task_id='extraction',
    dag=dag,
)

# 拟合任务（仅定义一次，对应当日的实例）
fitting = DummyOperator(
    task_id='fitting',
    dag=dag,
)

# 评估任务（仅定义一次，对应当日的实例）
eval_task = DummyOperator(
    task_id='eval',
    dag=dag,
)

2. 引用历史Task Instance设置依赖 #

Airflow中每个任务实例由task_id + execution_date唯一标识，我们可以通过dag.get_task(task_id).get_task_instance(target_date)来获取指定日期的任务实例，以此来设置跨日期的依赖：

# 计算昨日的执行日期
yesterday = dag.execution_date - timedelta(days=1)

# 设置拟合任务的依赖：当日抽取 + 昨日抽取
fitting.set_upstream([
    extraction,  # 引用当日的extraction实例
    dag.get_task('extraction').get_task_instance(yesterday)  # 引用昨日的extraction实例
])

# 设置评估任务的依赖：昨日拟合 + 当日抽取
eval_task.set_upstream([
    dag.get_task('fitting').get_task_instance(yesterday),  # 引用昨日的fitting实例
    extraction  # 引用当日的extraction实例
])

关键说明 #

• 为什么不用depends_on_past=True？这个参数是让当前任务依赖自身的昨日实例（比如当日fitting依赖昨日fitting），但你需要的是跨任务的昨日实例（当日fitting依赖昨日extraction），所以手动引用更灵活。
• 确保DAG的start_date早于你要引用的历史日期，比如如果你的第一个DAG run是2018-05-18，那么start_date至少要设为2018-05-17，这样才能获取到昨日的任务实例。
• 测试时可以用airflow tasks test model_training_pipeline fitting 2018-05-18验证依赖是否正确，Airflow会检查昨日extraction和当日extraction是否已完成。

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