如何用Airflow 2.x Taskflow API声明无返回值任务的依赖关系？

先看示例代码：

@dag(
    dag_id="chaining_tests",
    start_date=pendulum.datetime(2023, 11, 1),
    schedule=None
)
def chaining_tests(): 
    
    @task
    def validate_parameters(params=None):
        ## 校验参数，不合法则抛出Airflow异常
        pass
    
    @task
    def load_candidates(params=None) -> list[int]:
        return [1,2,3,4]
    
    @task
    def transform_candidate(candidate: int, params=None) -> str:
        return candidate+1
    
    
    # 在这里声明依赖
    validate_parameters()
    candidates = load_candidates()
    transform_candidates = transform_candidate.expand(candidate=candidates)
    # 后续逻辑

chaining_tests()

当前Airflow WebUI中的依赖图如下：

我希望load candidates任务在validate parameters任务执行完成后再运行。

已尝试的方案 #

方案1：传递参数 #

# 在这里声明依赖
valid = validate_parameters()
candidates = load_candidates(valid)

这种方法可行，但不够优雅——我们向下游任务传递了一个并不打算使用的参数。

方案2：使用移位运算符 #

这是混合Airflow 1.x和2.x语法的朴素尝试。

# 在这里声明依赖
validate_parameters() >> candidates = load_candidates()

显然不可行。

# 在这里声明依赖
validate_parameters()
candidates = load_candidates()
    
validate_parameters >> load_candidates

不被允许，报错：TypeError: unsupported operand type(s) for >>: '_TaskDecorator' and '_TaskDecorator'

candidates = load_candidates() << validate_parameters()
results = transform_candidate.expand(candidate=candidates)

抛出异常，报错：airflow.exceptions.XComForMappingNotPushed: did not push XCom for task mapping

方案3：借助XCom Magic #

该方案来自Stack Overflow的某个回答，但其工作原理尚不明确。

result1 = validate_parameters()  # 此任务无返回值
result2 = load_candidates()
result3 = transform_candidate.expand(candidate=result2)
result1 >> result2 >> result3

这种方法可行，但我们实际上是基于Airflow返回值定义依赖，而非直接定义任务间依赖，仍感觉像是一种取巧方式。而且这依旧是旧语法，而Airflow文档中提到：

相比之下，在Airflow 2.0的TaskFlow API中，任务调用本身会自动生成依赖关系

方案4：“Python化”方式 #

if validate_parameters(): 
   candidates = load_candidates() 
   results = transform_candidate.expand(candidate=candidates)

这里逻辑上“load_candidates”任务明显依赖于“validate_parameters”任务，但Airflow并不认可，依赖图回到初始状态。

问题 #

是否存在仅使用Airflow 2.x新Taskflow语法来声明无返回值任务间依赖的方法？
若不存在，针对不具备深厚Airflow知识的开发者，最佳实践是什么？

解答 #

1. 纯TaskFlow语法声明无返回值任务依赖的正确方式 #

TaskFlow API中，任务调用后返回的是任务实例对象，你可以直接对这些实例使用移位运算符>>或<<来声明依赖，这完全符合TaskFlow的设计逻辑，并非“旧语法”——移位运算符是Airflow 2.x中兼容TaskFlow的标准依赖声明方式：

@dag(
    dag_id="chaining_tests",
    start_date=pendulum.datetime(2023, 11, 1),
    schedule=None
)
def chaining_tests(): 
    
    @task
    def validate_parameters(params=None):
        ## 校验参数，不合法则抛出Airflow异常
        pass
    
    @task
    def load_candidates(params=None) -> list[int]:
        return [1,2,3,4]
    
    @task
    def transform_candidate(candidate: int, params=None) -> str:
        return str(candidate+1)
    
    
    # 正确声明依赖
    validate_task = validate_parameters()
    load_task = load_candidates()
    transform_tasks = transform_candidate.expand(candidate=load_task)
    
    # 用移位运算符定义依赖链
    validate_task >> load_task >> transform_tasks

chaining_tests()

这里的核心是：必须将任务调用的结果（任务实例）赋值给变量，再对变量使用移位运算符，而不是直接对装饰器对象（validate_parameters这种未调用的装饰器）操作——你之前方案2的错误就在于直接用了装饰器而非任务实例。

此外，TaskFlow还支持通过set_downstream()/set_upstream()方法声明依赖，效果和移位运算符一致：

validate_task.set_downstream(load_task)
load_task.set_downstream(transform_tasks)

2. 针对新手开发者的最佳实践 #

• 优先使用任务实例的移位运算符：这是Airflow 2.x中最直观且官方推荐的依赖声明方式，无论是有返回值还是无返回值的任务都适用，既符合TaskFlow的设计，又保持代码可读性。
• 避免无意义的参数传递：像方案1那样为了依赖而传递无用参数，会增加代码维护成本，还可能导致不必要的XCom数据存储，完全不推荐。
• 理解TaskFlow的“自动依赖”逻辑：文档中提到的“调用自动生成依赖”，指的是当任务A的返回值作为参数传入任务B时，自动建立A→B的依赖。如果任务无返回值，就需要手动通过移位运算符声明，这是TaskFlow的正常工作模式，并非“取巧”。
• 不要用Python条件语句模拟依赖：Airflow的DAG是在解析阶段生成的，Python的if语句只会影响DAG的结构生成，不会转化为任务间的执行依赖，方案4的方式完全不符合Airflow的运行逻辑。

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