Airflow 2.0动态生成任务分支上游任务无故被跳过问题排查

你的DAG出现任务无故跳过的核心原因是动态任务映射的调度时序与静态任务的执行时序不匹配：

• create_destination_raw_table是无依赖的静态任务，会在DAG启动后立即执行并快速完成；
• 而update_prestaging_table_columns及其上游的动态任务实例，需要等待get_s3_keys输出结果后才会被Airflow调度器创建；
• 当create_destination_raw_table完成时，insert_into_raw_table的其中一个依赖已满足，但动态任务分支的实例还未生成，Airflow 2.4.x的调度逻辑会误判该任务的依赖状态，进而跳过update_prestaging_table_columns及其下游任务。

另外，你用EmptyOperator重构DAG时正常，是因为静态任务和动态任务分支的执行时长差异被抹平，调度器有足够时间生成动态任务实例，不会触发误判。

方案1：调整静态任务的依赖关系（推荐） #

让create_destination_raw_table依赖get_s3_keys，确保静态任务不会在动态任务实例生成前完成，对齐两个分支的调度时序：

# 修改DAG依赖编排部分代码
with DAG(
    dag_id="random_skipping_DAG",
    description="Load JSON datasets from an S3 folder into a Redshift Database",
    start_date=datetime(2022, 12, 19),
    catchup=True,
    max_active_runs=1,
) as dag:

    s3_key_sensor = EmptyOperator(
        task_id="sense_daily_files",
    )

    get_s3_keys = PythonOperator(
        task_id="get_s3_keys",
        python_callable=lambda: ["key1", "key2", "key3"]
    )

    s3_key_sensor >> get_s3_keys

    # 让静态任务依赖get_s3_keys，确保动态任务实例开始生成后再执行
    destination_raw_table = create_destination_raw_table(
        schema=DESTINATION_SCHEMA,
        table_name=DESTINATION_TABLE,
    )
    get_s3_keys >> destination_raw_table

    # 动态任务分支保持原逻辑
    prestaging_tables = create_prestaging_redshift_table.partial(dest_schema=DESTINATION_SCHEMA).expand(s3_uri=get_s3_keys.output)
    prestaging_tables = load_s3_into_prestaging_table.expand_kwargs(prestaging_tables)
    prestaging_tables = update_prestaging_table_columns.expand_kwargs(prestaging_tables)

    prestaging_tables = insert_into_raw_table.partial(dest_table=destination_raw_table).expand_kwargs(prestaging_tables)

    drop_prestaging_tables.expand_kwargs(prestaging_tables)

方案2：升级Airflow版本 #

Airflow 2.4.x存在部分动态任务映射的调度bug，升级到2.5.0及以上版本可以修复此类依赖时序导致的任务跳过问题。

方案3：调整触发规则（辅助验证） #

如果暂时无法调整依赖或升级版本，可以为insert_into_raw_table设置触发规则为all_success（默认），同时确保动态任务分支的所有实例都能被正确生成：

@task(retries=10, retry_delay=60, trigger_rule="all_success")
def insert_into_raw_table(source_table: str, dest_table: str, **kwargs: Any) -> Dict[str, str]:
    # 原任务逻辑不变
    return {"table": source_table}

1. 部署修改后的DAG；
2. 触发一次新的调度运行，观察update_prestaging_table_columns是否正常执行，而非被跳过；
3. 确认insert_into_raw_table会在两个分支的依赖都完成后才启动。

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