如何在Kubeflow Pipeline中处理失败步骤并继续执行流水线？求可行实现示例

我刚好处理过类似的需求，Kubeflow Pipeline 默认会在某个步骤失败时跳过后续依赖它的任务，但我们可以通过任务的 status 属性结合 dsl.Condition 来实现「不管前一步成功失败都继续执行」的逻辑，完全符合你想要的伪代码效果。

核心思路：利用任务的status属性覆盖默认行为 #

每个 Kubeflow Pipeline 任务都有一个 status 属性，可能的取值包括 Succeeded（成功）、Failed（失败）、Skipped（跳过）等。默认情况下，任务失败后，后续依赖它的任务会被标记为 Skipped；我们可以通过条件判断「前一个任务没有被跳过」（即已经执行过，不管成功失败），强制后续任务运行。

基础实现代码 #

下面是对应你伪代码的具体实现：

import kfp
from kfp import dsl

# 定义几个测试组件：一个会失败的组件，一个成功的组件
@dsl.component
def failing_step():
    raise Exception("这是一个故意失败的步骤")

@dsl.component
def normal_step(step_name: str):
    print(f"执行 {step_name} 成功")

@dsl.pipeline(name="continue-on-fail-demo")
def pipeline():
    # 第一步：执行可能失败的任务
    step1 = failing_step()
    
    # 不管step1成功还是失败，只要它没被跳过就执行step2
    with dsl.Condition(step1.status != "Skipped"):
        step2 = normal_step(step_name="step2")
        
        # 同理，不管step2成功还是失败，执行step3
        with dsl.Condition(step2.status != "Skipped"):
            step3 = normal_step(step_name="step3")

运行这个流水线后，你会看到：即使 step1 失败，step2 和 step3 依然会正常执行，完全符合你的需求。

扩展：针对失败场景添加特殊处理 #

如果需要在某个步骤失败时执行额外的补救逻辑，还可以在条件中检查任务的失败状态，比如：

@dsl.pipeline(name="continue-on-fail-with-remediation")
def pipeline_with_remediation():
    step1 = failing_step()
    
    with dsl.Condition(step1.status != "Skipped"):
        step2 = normal_step(step_name="step2")
        
        # 如果step1失败，执行补救步骤
        with dsl.Condition(step1.status == "Failed"):
            remediation = normal_step(step_name="step1-remediation")
            # 确保补救步骤在step2之后执行
            remediation.after(step2)
        
        with dsl.Condition(step2.status != "Skipped"):
            step3 = normal_step(step_name="step3")

这个扩展方案中，step1 失败后，不仅 step2 会继续执行，还会额外触发一个补救步骤，兼顾了「继续执行」和「失败处理」的需求。

注意事项 #

• 确保条件判断的是 status != "Skipped"，而不是直接判断成功或失败，这样能覆盖所有前任务已执行的场景。
• 如果你的流水线使用了依赖关系（比如 step2.after(step1)），不需要删除这个依赖，条件判断会自动覆盖默认的失败跳过逻辑。

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