BranchPythonOperator调用后create_cluster_task无法触发的问题排查

你的create_cluster_task始终无法启动的核心原因是多Branch任务并行执行时的状态冲突：

1. 三个BranchPythonOperator是并行执行的，未检测到文件的Branch任务会先将create_cluster_task标记为skipped状态（因为它们的返回值中不包含该任务ID）。
2. 当后续检测到文件的Branch任务返回create_cluster_task时，该任务已经被标记为skipped，Airflow不会修改已确定的任务状态，因此无法启动执行。
3. 即使设置了one_success触发规则，由于部分上游Branch任务已将create_cluster_task标记为skipped，任务的整体状态判定仍会被跳过。

方案1：调整依赖结构（推荐） #

将create_cluster_task的上游改为各个文件移动任务，而非直接关联到Branch任务，从根源避免状态冲突：

修改代码中的依赖关系

# 调整后的依赖
check_cm_files_present_task >> move_to_valid_cm >> create_cluster_task
check_cm_files_present_task >> exit_task

check_cmts_struct_files_present_task >> move_to_valid_cmts_struct >> create_cluster_task
check_cmts_struct_files_present_task >> exit_task

check_cmts_meas_files_present_task >> move_to_valid_cmts_meas >> create_cluster_task
check_cmts_meas_files_present_task >> exit_task

create_cluster_task >> create_dataproc_cluster

修改Branch任务的返回值

每个Branch任务只需返回对应的移动任务或退出任务：

def check_cm_files_present():
    regex = re.compile(r"(.*?).gz")
    blobs = bucket.list_blobs(prefix=PREFIX)
    jsons = [blob for blob in blobs if regex.match(blob.name)]

    if len(jsons) > 0:
        return ["move_to_valid_cm"]  # 仅返回对应移动任务
    else:
        return ["exit_task"]

这样，只要任意一个移动任务执行成功，就会触发create_cluster_task，且one_success触发规则会确保只要有一个上游移动任务完成，它就执行，不受其他跳过任务的影响。

方案2：拆分独立集群创建DAG #

如果需要确保集群仅创建一次，可以将集群创建逻辑拆分为独立DAG，通过TriggerDagRunOperator触发：

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

def check_cm_files_present(**context):
    regex = re.compile(r"(.*?).gz")
    blobs = bucket.list_blobs(prefix=PREFIX)
    jsons = [blob for blob in blobs if regex.match(blob.name)]

    if len(jsons) > 0:
        # 触发独立的集群创建DAG
        TriggerDagRunOperator(
            task_id="trigger_cluster_creation",
            trigger_dag_id="create-dataproc-cluster",  # 替换为你的集群DAG ID
            dag=context["dag"]
        ).execute(context)
        return ["move_to_valid_cm"]
    else:
        return ["exit_task"]

这种方式彻底避免了多Branch任务对同一任务的状态干扰，确保集群创建逻辑独立执行。

方案3：调整触发规则（临时 workaround） #

如果坚持原有依赖结构，可以将create_cluster_task的触发规则改为none_failed_or_skipped，该规则要求所有上游任务无失败（允许成功或跳过）：

create_cluster_task = DummyOperator(
    task_id='create_cluster_task',
    trigger_rule='none_failed_or_skipped'
)

但此方案仍可能存在并行执行时的状态覆盖问题，仅作为临时替代方案。

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