Airflow中实现可选任务跳过且不阻塞下游任务的方案咨询

你当前的实现中，当should_run返回False时，do_run会被标记为skipped，而Task Group的状态会继承内部任务的最终状态（存在skipped任务时，Group状态为skipped）。下游next_task默认触发规则是all_success，因此会因为上游Group状态为skipped而被跳过。

通过BranchPythonOperator创建分支逻辑，让Task Group始终有一个成功完成的任务，确保Group状态为success，从而不影响下游任务执行。同时被跳过的任务会保留skipped状态颜色，完全符合你的需求。

修改后的代码如下：

from airflow import DAG
from airflow.models.param import Param
from airflow.operators.branch import BranchPythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.decorators import task, task_group

with DAG(
    dag_id="my_dag",
    params={"should_run": Param(False)}
):
    @task_group
    def my_group():
        # 分支逻辑：根据参数选择执行目标任务
        def should_run_branch(**kwargs):
            return "do_run" if kwargs["params"]["should_run"] else "skip_marker"

        branch_op = BranchPythonOperator(
            task_id="should_run_branch",
            python_callable=should_run_branch
        )

        @task
        def do_run():
            print("This is the optional task!")

        # 标记跳过的占位任务，始终成功完成
        skip_marker = DummyOperator(task_id="skip_marker")
        # Airflow 2.4+ 可使用更简洁的空任务替代：
        # @task.empty(task_id="skip_marker")()

        # 构建分支依赖
        branch_op >> [do_run, skip_marker]

    @task()
    def next_task():
        print("Next task, other tasks can follow this one")

    my_group() >> next_task()

• 当should_run为True时，分支到do_run执行，skip_marker会被标记为skipped；
• 当should_run为False时，分支到skip_marker执行，do_run会被标记为skipped（显示对应状态颜色）；
• 无论哪种情况，Task Group内都有一个任务成功完成，因此Group状态为success，下游next_task会按默认规则正常执行；
• 无需修改下游任务触发规则，也没有在业务任务do_run内部编写判断逻辑。

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