如何在Airflow多分支流程中复用同一个任务？

完全可以复用Airflow中多分支共用的任务，你的代码思路本身具备可行性，但需要注意细节调整，同时还有更优雅的实现方式：

一、修正你的现有代码 #

你的代码存在拼写错误：task2应改为task_2，修正后即可正常运行。此时task_comm的执行逻辑是所有上游任务（task_2和task_3）都成功完成后才会启动，这是Airflow的默认触发规则。

修正后的完整代码：

from airflow.operators.dummy import DummyOperator
# 假设branch是已定义的分支任务，比如BranchPythonOperator等

flow_1 = DummyOperator(task_id='flow_1')
task_1 = DummyOperator(task_id='task_1')
task_2 = DummyOperator(task_id='task_2')

flow_2 = DummyOperator(task_id='flow_2')
task_3 = DummyOperator(task_id='task_3')

task_comm = DummyOperator(task_id='task_comm')

branch >> flow_1 >> task_1 >> task_2 >> task_comm
branch >> flow_2 >> task_3 >> task_comm

二、更优雅的复用方式 #

如果task_comm包含复杂逻辑，或需要在多个DAG中复用，推荐以下两种方式：

1. 封装为可复用函数 #

将任务创建逻辑封装成函数，方便在任意位置调用：

from datetime import timedelta
from airflow.operators.dummy import DummyOperator

def get_common_task():
    return DummyOperator(
        task_id='task_comm',
        owner='airflow',
        retries=2,
        retry_delay=timedelta(minutes=5)
        # 添加其他通用配置
    )

# 在流程中调用
task_comm = get_common_task()

2. 自定义Operator（适合复杂业务逻辑） #

如果通用任务有专属的业务逻辑，可自定义Operator实现复用：

from airflow.models.baseoperator import BaseOperator
from airflow.utils.decorators import apply_defaults
from datetime import timedelta

class CommonTaskOperator(BaseOperator):
    @apply_defaults
    def __init__(self, retry_delay=timedelta(minutes=3), **kwargs):
        super().__init__(**kwargs)
        self.retry_delay = retry_delay

    def execute(self, context):
        # 这里编写通用任务的具体业务逻辑
        self.log.info("执行通用任务：处理分支流程的收尾工作")
        # 示例逻辑：可在这里调用API、处理数据等

# 使用自定义Operator
task_comm = CommonTaskOperator(task_id='task_comm', retries=2)

三、调整触发规则适配不同场景 #

根据业务需求，可修改task_comm的触发规则：

• 默认规则（所有上游完成）：如你的原始逻辑，需两条分支都执行完才启动task_comm
• 任意分支完成即执行：设置trigger_rule=TriggerRule.ONE_SUCCESS，只要有一条分支完成就启动：

  from airflow.utils.trigger_rule import TriggerRule
  from airflow.operators.dummy import DummyOperator
  
  task_comm = DummyOperator(
      task_id='task_comm',
      trigger_rule=TriggerRule.ONE_SUCCESS
  )
  

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