如何在Airflow Taskflow API中合理混用有/无返回值任务？

常见场景写法 #

• 无返回值任务的串行实现很直接：

  t1() >> t2()
  

• 任务有返回值且后续任务需要使用时，写法如下：

  return_value = t1()
  t2(return_value)
  

问题场景 #

当需要混用两种情况——比如要实现t1 >> t2 >> t3的串行，其中t2有返回值要传给t3，同时t1必须在t2之前执行——直接按直觉写会出问题：

• 下面的代码会触发语法错误，因为>>运算符不能用于有返回值的任务赋值语句前：

  t1() >> 
  returned_value = t2()
  t3(returned_value)
  

• 如果去掉>>写成下面这样，t1会和t2、t3并行运行，达不到串行要求：

  t1()
  returned_value = t2()
  t3(returned_value)
  

• 一种不太优雅的临时解决办法是让t2接收t1的返回值（即便逻辑上不需要）：

  returned_fake_t1 = t1()
  returned_value_t2 = t2(returned_fake_t1)
  t3(returned_value_t2)
  

  但这种方式需要修改任务逻辑，不够合理。

惯用优雅实现方式 #

Taskflow API提供了两种更合适的方式来处理这种场景：

1. 利用海象运算符:=同时建立依赖和保留返回值引用
   通过海象运算符在建立串行依赖的同时，获取t2的任务对象，后续通过task.output获取返回值：

   t1() >> (t2_obj := t2())
   t3(t2_obj.output)
   

   这种写法既保证了t1 >> t2 >> t3的串行依赖，又不需要修改t2的参数逻辑。

2. 显式调用任务的set_upstream方法
   如果不习惯海象运算符，可以显式设置任务的上游依赖，再通过任务对象的output属性获取返回值：

   t1_task = t1()
   t2_task = t2()
   t2_task.set_upstream(t1_task)
   t3(t2_task.output)
   

   同样能实现预期的串行DAG结构，且无需修改任务本身的逻辑。

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