Python 脚本一崩就得重跑整夜？用 Prefect 给数据任务加上重试和缓存

写过数据脚本的人大概都经历过这个：一个跑三小时的批处理任务，在第 2 小时 50 分崩了。你修完 bug 重跑，又要三小时。更糟的是脚本里没有断点，重跑意味着从头再来。

Prefect 想解决的正是这类"脚本很脆、重跑很贵"的问题。它在 GitHub 上有 2.4 万 star，Apache-2.0 协议，商用没问题。它的思路和 Airflow 很不一样——不是让你把脚本改写成 DAG，而是让你给现有 Python 函数加两个装饰器，立刻获得重试、缓存、可观测性。

它到底是什么

Prefect 是一个用 Python 写的数据流水线编排框架。核心卖点写在它的定位里：构建"有弹性的"（resilient）数据管道。

和 Airflow 最大的区别在于侵入性。Airflow 要求你按它的 DAG 范式重写工作流；Prefect 的做法是：你正常写 Python 函数，加个 @task 或 @flow 装饰器，它就有了编排能力。对已经有大量 Python 脚本的团队，这个迁移成本几乎为零。

三个概念要分清：

• Flow：整个工作流，一个 Python 函数
• Task：工作流里的一个步骤
• Result：任务的输出，可以被持久化、被缓存

缓存是 Prefect 特别好用的一点：一个任务如果输入没变，重跑时可以直接复用上次的结果，跳过执行。对那些"前面几步很贵、只改了最后一步"的场景，这个特性能省下大量时间。

装饰器怎么用

一个最小的例子：

from prefect import flow, task

@task(retries=3, retry_delay_seconds=60)

def extract():

return fetch_data()

@task(cache_key_fn=lambda *a, **k: "transform-v1")

def transform(data):

return clean(data)

@flow(name="daily-pipeline")

def pipeline():

raw = extract()

clean = transform(raw)

pipeline()

这个例子里藏了三个关键能力：

• retries=3——失败自动重试三次
• retry_delay_seconds=60——重试间隔，配合退避策略避免把下游服务打垮
• cache_key_fn——给任务结果一个缓存键，输入不变就跳过

你没写一行调度逻辑，也没画任何图，就有了重试和缓存。这就是 Prefect 的哲学：编排能力应该是 Python 的一部分，而不是另一套系统。

什么情况选 Prefect，什么情况选 Airflow

这是被问得最多的问题。我的判断标准是这样的：

选 Prefect 的情况：

• 团队主力是 Python，脚本已经写了一大堆，不想重写
• 需要精细的重试、缓存、超时控制
• 工作流的动态性强——根据上一步的结果决定下一步做什么
• 想让"跑一次"和"定时跑"用同一套代码

选 Airflow 的情况：

• 团队里有非 Python 同学也要维护流水线（SQL、Bash 居多）
• 需要强大的 Web UI 做运维和人工干预
• 生态插件依赖重（Airflow 的 provider 生态目前还是最全的）
• 需要复杂的回填和 SLA 管理

一句话：Prefect 更"Pythonic"，适合工程团队把脚本工业化；Airflow 更"平台化"，适合当全公司的调度中心。

部署的两种路

Prefect 可以完全本地跑，一行命令起一个服务端：

prefect server start

它会在本地起一个 UI，你能看到每次运行的状态、耗时、日志。对个人和小团队，这就够了，不用上云。

要团队协作或生产部署，可以用 Prefect Cloud，也可以自建。自部署的关键是把数据库从默认的 SQLite 换成 Postgres，否则并发一上来就锁死。

调度本身很简单——给 flow 加个参数：

@flow

def pipeline(): ...

if __name__ == "__main__":

pipeline.serve(name="daily", cron="0 2 * * *")

上手要避开的三个坑

别在 task 之间传大对象。 和 Airflow 一样，Prefect 的中间结果默认会持久化到数据库。把一个几百 MB 的 DataFrame 当 task 返回值，会把元数据库拖垮。大对象走外部存储，task 之间只传路径。

重试不是万能药。 有些失败重试一百次也不会成功——比如数据格式不对、字段缺失。这类错误应该直接失败并告警，而不是靠重试掩盖。给不同类型的异常配置不同的重试策略，这是保证流水线可信的关键。

并发要显式限制。 默认配置下 Prefect 会尽量并行跑任务，如果你的下游是个扛不住并发的老系统，记得加并发标签，别把人家打挂。

让流水线真正好用的四个实践

用 tags 做任务分类治理。 给 task 打标签（tags={"db"}、tags={"llm"}），之后可以按标签设并发限制、按标签过滤 UI。任务多了以后，没有分类的流水线会变成一团乱麻。

日志要结构化。 在 task 里用 get_run_logger() 而不是 print()，日志会自动关联到这次运行和这个任务。排查问题时能直接定位到"哪次运行的哪一步"，不用在一大段 stdout 里翻。

from prefect import task, get_run_logger

@task

def risky_step():

logger = get_run_logger()

logger.warning("数据行数异常: %s", row_count)

参数化，别硬编码日期。 flow 接受参数，运行的时候传。这样手动重跑某一天、或者补跑某个区间，都只是改个参数的事，不用改代码。

给关键流程加告警。 用 on_failure 钩子挂通知，失败立刻推到 Slack 或飞书。监控的价值在于"第一时间知道"，事后再发现意义就减半了。

成本与资源的现实考量

自部署 Prefect 的资源开销很小，一个 2 核 4G 的服务器跑日常量的流水线完全够。真正的成本在数据库：每一次运行、每一个任务的状态都会写进去，跑得频繁的话几个月能涨到几十 GB。

建议提前做两件事：配置数据库的数据保留策略（自动清理 N 天前的历史），以及把大对象的持久化关掉（persist_result=False）。这两项不做，半年后数据库会成为新的故障点。

我的判断

Prefect 真正的价值不在"调度"，而在"让普通 Python 脚本具备生产级韧性"。大部分团队其实不需要一套重型调度平台，他们需要的只是：脚本崩了能自动重试、能知道崩在哪、重跑不用从头来。这三件事，Prefect 用几个装饰器就给了你。

如果你手上正有一堆"能跑但不敢交给它"的 Python 脚本，这是目前改造成本最低的一条路。

官方仓库在 github.com/PrefectHQ/prefect。