DataprocCreateClusterOperator的one_success触发规则失效问题

首先看你代码里的语法错误，这可能是导致任务异常的根本原因：

每个BashOperator的bash_command都存在引号未闭合、命令末尾多逗号的问题，比如：

move_to_valid_cm = BashOperator(
    task_id='move_to_valid_cm',
    bash_command='''gsutil -m mv gs://A gs://B,  # 多了逗号且未闭合三个单引号
    dag = dag
)

这种错误会导致BashOperator执行时命令无效，任务直接失败。如果所有move_to_xxx任务都失败，create_dataproc_cluster的one_success规则自然无法触发——毕竟没有任何上游任务成功。

针对trigger_rule不生效的核心解决方案 #

假设你已经修正了语法错误，仍存在one_success不触发的情况，可按以下步骤排查：

1. 替换trigger_rule为none_failed_min_one_success #

one_success规则允许上游存在失败任务（只要至少一个成功就触发），但你的场景中上游任务只会是成功或跳过（skipped）（由BranchPythonOperator分支逻辑决定），更适合用none_failed_min_one_success：

create_dataproc_cluster = dataproc_operator.DataprocCreateClusterOperator(
    task_id='create_dataproc_cluster',
    cluster_name=CLUSTER_WITH_DATE,
    project_id=PROJECT,
    cluster_config=CLUSTER_CONFIG,
    region='europe-west3',
    trigger_rule='none_failed_min_one_success'  # 替换为该规则
)

这个规则要求：所有上游任务都未失败（成功或跳过均可），且至少有一个上游任务成功，完全匹配你的业务场景。

2. 验证BranchPythonOperator的分支逻辑 #

确保你的check_xxx函数返回的task_id完全正确：

• 当文件存在时，返回对应move_to_xxx的task_id，确保该move任务被执行（状态为success）
• 当文件不存在时，返回exit_task的task_id，确保move任务被标记为skipped

可以在Airflow UI的任务实例详情中查看每个move任务的状态，确认是否有success的任务存在。

3. 检查Airflow版本兼容性 #

如果使用Airflow 1.x版本，trigger_rule的处理存在一些已知bug，比如对skipped状态的任务判断逻辑异常。建议升级到Airflow 2.x版本，新版本对trigger_rule的处理更准确。

4. 调整任务依赖结构（可选） #

如果上述方法无效，可以添加中间DummyOperator作为过渡，将每个move任务连接到一个DummyOperator，再将这些DummyOperator作为create_dataproc_cluster的上游：

from airflow.operators.dummy import DummyOperator

dummy_cm = DummyOperator(task_id='dummy_cm', dag=dag)
dummy_cmts_struct = DummyOperator(task_id='dummy_cmts_struct', dag=dag)
dummy_cmts_meas = DummyOperator(task_id='dummy_cmts_meas', dag=dag)

move_to_valid_cm >> dummy_cm
move_to_valid_cmts_struct >> dummy_cmts_struct
move_to_valid_cmts_meas >> dummy_cmts_meas

[dummy_cm, dummy_cmts_struct, dummy_cmts_meas] >> create_dataproc_cluster

这种方式可以避免skipped状态直接传递到create_dataproc_cluster的上游，可能解决规则判断异常的问题。

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