Airflow DAG条件任务执行问题：如何按需跳过指定任务

你的问题核心在于：Airflow的DAG是在解析阶段生成任务依赖结构的，直接用Python的if判断任务返回值是行不通的——因为解析阶段users_from_b、uniq_users都是Task对象，不是实际运行时的返回数据，所以你的条件判断逻辑根本没在正确的时机执行。

下面是两种可行的实现方式：

方案一：使用ShortCircuitOperator控制任务流 #

ShortCircuitOperator会根据指定的判断逻辑，决定是否让下游任务继续执行。如果判断返回False，下游所有依赖该任务的任务都会被标记为跳过。

修改后的完整代码：

from airflow import DAG
from airflow.operators.python import PythonOperator, ShortCircuitOperator
from datetime import days_ago

DAG_SCHEDULE = "@daily"  # 替换为你的实际调度间隔

@dag(
    dag_id="id_123",
    schedule=DAG_SCHEDULE,
    start_date=days_ago(0),
    catchup=False,
    default_args={
        "retries": 0,
    },
)
def dag_runner():
    @task(task_id="get_data_src_a")
    def get_data_src_a() -> list:
        # return data from src_a
        return []

    @task(task_id="get_data_src_b")
    def get_data_src_b() -> list:
        # return data from src_b
        return []

    @task(task_id="find_uniq_users")
    def find_uniq_users(users_from_a, users_from_b) -> list:
        # return users in src_a but not in src_b
        return list(set(users_from_a) - set(users_from_b))

    @task(task_id="do_something_with_users")
    def do_something_with_users(uniq_users):
        # do something with unique users
        pass

    # 定义判断是否执行find_uniq_users的逻辑
    def should_run_find_uniq(users_from_b):
        return len(users_from_b) > 0

    # 定义判断是否执行do_something的逻辑
    def should_run_do_something(uniq_users):
        return len(uniq_users) > 0

    users_from_a = get_data_src_a()
    users_from_b = get_data_src_b()

    # 第一个短路判断：控制find_uniq_users是否执行
    short_circuit_find = ShortCircuitOperator(
        task_id="short_circuit_find_uniq",
        python_callable=should_run_find_uniq,
        op_kwargs={"users_from_b": users_from_b}
    )

    uniq_users = find_uniq_users(users_from_a, users_from_b)
    short_circuit_find >> uniq_users

    # 第二个短路判断：控制do_something_with_users是否执行
    short_circuit_do = ShortCircuitOperator(
        task_id="short_circuit_do_something",
        python_callable=should_run_do_something,
        op_kwargs={"uniq_users": uniq_users}
    )

    do_something_task = do_something_with_users(uniq_users)
    short_circuit_do >> do_something_task

dag_runner()

方案二：在任务内部抛出SkipException #

如果不想额外增加控制任务，可以在目标任务内部判断条件，当需要跳过的时候抛出airflow.exceptions.SkipException，Airflow会自动将该任务标记为跳过，下游依赖任务（默认trigger_rule为all_success）也会被跳过。

修改后的关键代码：

from airflow.exceptions import SkipException

@task(task_id="find_uniq_users")
def find_uniq_users(users_from_a, users_from_b) -> list:
    if not users_from_b:
        raise SkipException("users_from_b is empty, skip this task")
    # return users in src_a but not in src_b
    return list(set(users_from_a) - set(users_from_b))

@task(task_id="do_something_with_users")
def do_something_with_users(uniq_users):
    if not uniq_users:
        raise SkipException("uniq_users is empty, skip this task")
    # do something with unique users
    pass

# 任务依赖保持原有结构
users_from_a = get_data_src_a()
users_from_b = get_data_src_b()
uniq_users = find_uniq_users(users_from_a, users_from_b)
do_something_with_users(uniq_users)

这种方式更简洁，直接在业务任务内部处理跳过逻辑，适合逻辑简单的场景。

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