3、Task状态管理：状态机模型

Task 状态管理，说白了就是 Prefect 工作流的心脏。我刚开始接触 Prefect 时，觉得状态不就是「成功」或「失败」嘛，有啥好研究的？直到有一次线上调度出了大问题——一个任务卡在 Pending 状态整整 8 小时，下游任务全等着，整个数据管道瘫痪了。嗯，从那以后我才真正重视起状态机模型。

Prefect 的每个 Task 都遵循一套严格的状态机模型。状态不是随便跳的，它有固定的转换规则。你想想看，如果任务能从 Running 直接跳到 Success，那中间要是出错了怎么办？所以 Prefect 设计了一套完整的状态流转体系。

3.1 核心状态一览

Prefect 定义了 8 种核心状态，我习惯把它们分成三组：

状态 含义 是否终态
Pending 任务已创建，等待调度 否
Running 任务正在执行 否
Success 任务执行成功 是
Failed 任务执行失败 是
Cancelled 任务被取消 是
Crashed 任务异常崩溃 是
Scheduled 任务已排期，等待未来执行 否
Retrying 任务正在重试 否

这里面有个细节：Scheduled 其实是 Pending 的子状态。我个人习惯把 Scheduled 理解为「带时间戳的 Pending」——它知道什么时候该跑，而普通 Pending 只是等着被调度器分配。

核心原则：非终态可以继续流转，终态一旦到达就不会再改变。Success、Failed、Cancelled、Crashed 都是终态，任务到了这些状态就「盖棺定论」了。

3.2 状态转换规则

状态转换不是随意的。Prefect 内部维护了一张严格的状态转换表。我画了张图，你看一眼就明白了：

Prefect Task 状态机流转图 Prefect Task 状态机流转图 Pending Scheduled Running Retrying Success Failed Cancelled Crashed schedule run time reached complete error retry retry cancel crash 说明： 实线箭头表示正常流转路径，虚线箭头表示重试后的回环路径。 终态（Success/Failed/Cancelled/Crashed）到达后不再向外转换。

这里有几个关键规则，我重点说一下：

  • Pending 可以双向走：既可以进入 Scheduled 等待排期，也可以直接进入 Running 立即执行。这取决于你用的是 .submit() 还是 .map()。
  • Running 是唯一的活动状态：只有 Running 状态的任务才在执行代码。其他状态要么在等，要么已经结束。
  • Retrying 是个「回环」：任务从 Running 进入 Retrying，然后重新回到 Running。这不是终态，它是个中间过渡。
  • Crashed 和 Failed 不一样：Failed 是业务逻辑抛出了异常，Crashed 是基础设施层面挂了（比如进程被 kill、内存溢出）。

我曾经踩过一个坑：在自定义状态回调里试图把 Failed 状态改成 Success。结果 Prefect 直接报错，说「不能从终态转换到非终态」。后来我才明白，状态机一旦到达终态，就彻底锁死了。想「复活」任务？只能重新提交一个新的 Task Run。

3.3 自定义状态

Prefect 允许你创建自定义状态。说实话，这个功能我用得不多，但一旦用上就特别爽。比如我有个数据质量校验任务，我想定义个 DataQualityWarning 状态——数据能跑通，但质量不达标。

from prefect.states import State

class DataQualityWarning(State):
    """数据质量警告状态"""
    def __init__(self, message: str = "", **kwargs):
        super().__init__(type="DATA_QUALITY_WARNING", message=message, **kwargs)
        self.quality_score = kwargs.get("quality_score", 0.0)

@task
def validate_data(df):
    quality = compute_quality(df)
    if quality < 0.8:
        # 返回自定义状态，不是 Failed
        return DataQualityWarning(
            message=f"数据质量评分 {quality}，低于阈值 0.8",
            quality_score=quality
        )
    return "数据质量合格"

你想想看，这样做的好处是什么？下游任务可以针对这个自定义状态做不同处理：

@task
def handle_quality_warning(state: DataQualityWarning):
    if state.quality_score > 0.6:
        # 勉强能用，发个告警就行
        send_alert(f"数据质量偏低: {state.message}")
    else:
        # 太差了，直接阻塞下游
        raise ValueError(f"数据质量太差: {state.message}")

我的建议：自定义状态不要滥用。Prefect 内置的 8 种状态覆盖了 95% 的场景。只有当你需要表达「业务语义」而非「执行语义」时，才考虑自定义状态。比如「数据质量警告」是业务语义，「任务被限流」其实可以用 Pending 表达。

3.4 状态回调函数

状态回调是 Prefect 最实用的功能之一。说白了，就是在状态发生变化时，自动触发一段代码。我个人习惯用回调来做三件事：日志记录、告警通知、数据持久化。

回调函数有两种注册方式：

方式一：Task 级别回调

from prefect import task
from prefect.states import State

def on_state_change(task, old_state: State, new_state: State):
    """状态变化时自动调用"""
    print(f"[回调] {task.name}: {old_state.type} -> {new_state.type}")
    if new_state.type == "FAILED":
        # 发企业告警
        send_wechat_alert(f"任务 {task.name} 失败了！")
    elif new_state.type == "SUCCESS":
        # 记录成功日志到数据库
        log_to_db(task.name, "success", new_state.result())

@task(on_state_change=on_state_change)
def my_task():
    return 42

方式二：Flow 级别全局回调

from prefect import flow

def global_state_handler(task, old_state, new_state):
    """全局状态监控"""
    if new_state.type in ("FAILED", "CRASHED"):
        # 记录到监控系统
        metrics.inc("task_failure", tags={"task": task.name})

@flow(task_runner=SequentialTaskRunner())
def my_flow():
    # 注册全局回调
    for t in [task_a, task_b, task_c]:
        t.on_state_change = global_state_handler
    
    task_a()
    task_b()
    task_c()

这里有个细节：回调函数的参数顺序是 task, old_state, new_state。我刚开始写反过，结果回调一直不触发，排查了半天才发现参数顺序错了。

实战经验：回调函数里不要做耗时操作。我曾经在回调里写了个数据库写入，结果任务本身 1 秒跑完，回调花了 3 秒。Prefect 的回调是同步执行的，回调没跑完，任务状态就不会最终确认。所以回调里尽量用异步或消息队列。

3.5 状态机实战：一个完整的例子

把上面这些知识点串起来，写一个带状态管理的 ETL 任务：

from prefect import task, flow
from prefect.states import State, Failed, Success
import time

class RetryLimitExceeded(State):
    """自定义状态：重试次数超限"""
    def __init__(self, retry_count: int, **kwargs):
        super().__init__(type="RETRY_LIMIT_EXCEEDED", **kwargs)
        self.retry_count = retry_count

def etl_state_callback(task, old_state, new_state):
    """ETL 状态监控回调"""
    print(f"[{time.strftime('%H:%M:%S')}] {task.name}: {old_state.type} -> {new_state.type}")
    
    if isinstance(new_state, Failed):
        # 记录失败原因
        with open("etl_errors.log", "a") as f:
            f.write(f"{task.name} 失败: {new_state.message}\n")
    
    if isinstance(new_state, RetryLimitExceeded):
        # 重试超限，发紧急告警
        send_pagerduty_alert(f"ETL 任务 {task.name} 重试 {new_state.retry_count} 次后放弃")

@task(
    retries=3,
    retry_delay_seconds=10,
    on_state_change=etl_state_callback
)
def extract_data(source: str):
    """数据抽取，带重试机制"""
    # 模拟可能失败的数据抽取
    if source == "unstable_api":
        raise ConnectionError("API 连接不稳定")
    return {"data": [1, 2, 3], "source": source}

@task(on_state_change=etl_state_callback)
def transform_data(raw_data: dict):
    """数据转换"""
    if not raw_data.get("data"):
        return RetryLimitExceeded(
            retry_count=3,
            message="数据为空，无法转换"
        )
    return [x * 2 for x in raw_data["data"]]

@flow
def etl_pipeline():
    raw = extract_data("unstable_api")
    transformed = transform_data(raw)
    return transformed

if __name__ == "__main__":
    result = etl_pipeline()
    print(f"最终状态: {result}")

这个例子展示了几个关键点：

  • 重试机制下，状态会在 Running -> Retrying -> Running 之间循环
  • 自定义状态 RetryLimitExceeded 在业务逻辑中手动返回
  • 回调函数统一记录状态变化，失败时写日志，重试超限时发告警

一个小技巧：在回调函数里，可以通过 new_state.result() 拿到任务的返回值。但要注意——只有 Success 状态才有 result，Failed 状态拿 result 会抛出异常。所以记得先判断状态类型。

状态管理这块，说白了就是给任务装了个「黑匣子」。每个状态变化都被记录，每个转换都有规则约束。我做了这么多年数据工程，见过太多因为状态管理混乱导致的事故——任务卡住了没人知道、失败了没有告警、重试逻辑写成了死循环。Prefect 的状态机模型把这些都规范化了，你只要按规则来，基本不会出大问题。

← 上一章 📖 返回目录 下一章 →