Airflow条件任务：使用@task.short_circuit实现动态跳过

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
import random

# 定义DAG调度，这里设置为None表示手动触发或不定期
DAG_SCHEDULE = None

@dag(
    dag_id="conditional_tasks_with_short_circuit",
    schedule=DAG_SCHEDULE,
    start_date=days_ago(0),
    catchup=False,
    default_args={
        "retries": 0,
    },
    tags=["example", "conditional"]
)
def dag_runner():
    @task(task_id="get_data_src_a")
    def get_data_src_a() -> list:
        """
        模拟从数据源A获取数据。
        随机返回一个非空列表或空列表。
        """
        if random.random() > 0.1: # 90%概率返回数据
            print("数据源A：获取到用户数据。")
            return ["user_a1", "user_a2", "user_a3"]
        print("数据源A：未获取到用户数据。")
        return []

    @task(task_id="get_data_src_b")
    def get_data_src_b() -> list:
        """
        模拟从数据源B获取数据。
        随机返回一个非空列表或空列表。
        """
        if random.random() > 0.5: # 50%概率返回数据
            print("数据源B：获取到用户数据。")
            return ["user_a1", "user_b1"]
        print("数据源B：未获取到用户数据。")
        return []

    @task(task_id="find_uniq_users")
    def find_uniq_users(users_from_a: list, users_from_b: list) -> list:
        """
        查找在数据源A中但不在数据源B中的唯一用户。
        """
        # 确保输入是列表，以防上游任务被跳过而导致None值
        users_from_a = users_from_a or []
        users_from_b = users_from_b or []

        uniq_users = [u for u in users_from_a if u not in users_from_b]
        print(f"找到的唯一用户: {uniq_users}")
        return uniq_users

    @task(task_id="do_something_with_users")
    def do_something_with_users(uniq_users: list):
        """
        对唯一用户执行某些操作。
        """
        print(f"正在处理唯一用户: {uniq_users}")
        # 模拟一些处理逻辑
        import time
        time.sleep(1)
        print("唯一用户处理完成。")

    # --- 使用 @task.short_circuit 实现条件逻辑 ---

    @task.short_circuit(task_id="should_find_uniq_users")
    def should_find_uniq_users(users_from_b: list) -> bool:
        """
        检查数据源B的用户列表是否为空。
        如果为空，则短路下游任务（find_uniq_users及其后续）。
        """
        if not users_from_b:
            print("条件判断：数据源B的用户列表为空，将跳过 'find_uniq_users'。")
            return False
        print("条件判断：数据源B的用户列表不为空，将继续执行 'find_uniq_users'。")
        return True

    @task.short_circuit(task_id="should_do_something_with_users")
    def should_do_something_with_users(uniq_users: list) -> bool:
        """
        检查唯一用户列表是否为空。
        如果为空，则短路下游任务（do_something_with_users）。
        """
        if not uniq_users:
            print("条件判断：唯一用户列表为空，将跳过 'do_something_with_users'。")
            return False
        print("条件判断：唯一用户列表不为空，将继续执行 'do_something_with_users'。")
        return True

    # 实例化任务
    users_from_a_result = get_data_src_a()
    users_from_b_result = get_data_src_b()

    # 第一个短路任务：检查users_from_b是否为空
    # should_find_uniq_users 任务依赖于 users_from_b_result 的输出
    check_b_data_task = should_find_uniq_users(users_from_b=users_from_b_result)

    # find_uniq_users 任务依赖于两个数据源的结果以及第一个短路任务的判断
    # 如果 check_b_data_task 返回 False，则 uniq_users_result 及其下游将被跳过
    uniq_users_result = find_uniq_users(users_from_a=users_from_a_result, users_from_b=users_from_b_result)

    # 设定依赖关系
    # 两个数据获取任务完成后，才能进行第一个条件判断
    # 并且 find_uniq_users 任务的执行受 check_b_data_task 控制
    [users_from_a_result, users_from_b_result] >> check_b_data_task >> uniq_users_result

    # 第二个短路任务：检查uniq_users是否为空
    # should_do_something_with_users 任务依赖于 uniq_users_result 的输出
    check_uniq_users_task = should_do_something_with_users(uniq_users=uniq_users_result)

    # do_something_with_users 任务依赖于 uniq_users_result 和第二个短路任务的判断
    # 如果 check_uniq_users_task 返回 False，则 do_something_with_users 将被跳过
    uniq_users_result >> check_uniq_users_task >> do_something_with_users(uniq_users=uniq_users_result)

dag_runner()