用Prefect实现Python工作流编排：从依赖管理到生产落地

如果你维护过超过二十个数据脚本大概率体会过这种日子任务之间的依赖靠执行顺序和文件名约定失败靠同事碰巧看到告警邮件重跑靠手工把断掉的那一段捡起来。问题不在脚本写得不好而在于没有人用工程化的方式管理“任务之间的状态和依赖”。这正是我后来认真研究 Prefect 的原因。它没有发明新的编排范式而是把 Python 工作流编排拉回了本质任务就是函数流程就是调用关系状态就是可查询的数据。这篇文章不是官方文档的翻译是我从选型、写任务、部署到生产踩坑的一整套复盘适合刚接触编排工具、又不想一上来就被调度器复杂度劝退的人。1. 从“一把梭脚本”到编排框架工作流到底缺了什么1.1 脚本长大的痛苦sleep、标志文件与手工重跑我待过的一个团队数据管道最早就是三个脚本抓数、清洗、入库。跑法也简单在 Shell 里按顺序执行每跑完一个再跑下一个。后来需求多了脚本从三个变成二十多个问题就来了每天要么靠人盯要么在 cron 里写sleep 300祈祷上游这五分钟能跑完。最痛的是“重跑”这件事。脚本之间没有真正的依赖描述只靠“上一个脚本写出的文件是否存在”来传递状态。一旦中间某一步失败修复之后想只跑失败的那一段和它的下游几乎不可能。最安全的办法就是把整条链路从头跑一遍耗时翻倍还不敢保证不会漏掉某个步骤。我当时下定决心做编排改造不是因为想追求什么先进架构而是因为我已经算不清“哪个任务依赖哪个任务”了。手工维护的“脑内 DAG”复杂度一旦超过十个节点就必然出错。后来我看到不少团队甚至用 Excel 表格维护任务依赖清单更加确信流程编排本身已经成了一件正经事靠人肉记是不可持续的。1.2 编排要解决的四件事依赖、状态、恢复、可见性把上面的痛苦抽象一下编排工具本质上要解决四件事无论它把自己包装成什么样的平台、产品或者框架问题没有编排时的典型症状依赖管理用 sleep、文件标记、执行顺序来隐式表达先后关系状态追踪不知道某任务当前是成功、失败、还是压根没到执行时间失败恢复重跑只能整条链路重来或者靠人肉记住从哪一步继续可观测性日志散落在不同目录参数、耗时、触发原因无从追溯我后来用这个标准去评价工具发现它可以解释很多选择。比如为什么有人宁可忍受 Airflow 的部署复杂度也不回退到 cron因为 Airflow 至少在状态和依赖上提供了可靠答案。为什么有人连 Airflow 都不想要因为他们觉得为几个任务搭一套 scheduler 加数据库太重了希望更轻的东西介入。Prefect 恰好站在“轻”和“严谨”的交界处它想用尽量少的运维成本把上面四件事都做扎实。顺带说一句如果你们任务只有三个串行跑五分钟结束别急着上框架。编排工具也是成本最简单的脚本加一个成体系的日志目录可能才是最优解。判断标准很简单当“漏掉一个任务”的后果开始让你失眠才真的需要它。1.3 为什么非要是 Python代码即流程流程即代码接着回答一个常被问的问题为什么工作流描述语言必须是 Python毕竟 YAML、JSON 可以描述任务列表、依赖关系和调度时间看起来不是更简单吗我的体验是声明式配置在“简单流程”很好用一旦流程开始出现分支、循环、动态参数配置文件的表达力会迅速崩溃。你会在 YAML 里写各种模板表达式你也得为“根据上一个任务返回值的长度决定要不要跑下游”这种逻辑设计一堆自定义关键字最后等于在造一门新方言。Prefect 的做法不一样它承认“任务本来就是 Python 函数流程本来就是函数之间的调用关系”所以干脆用装饰器把普通函数升级成可追踪的执行单元。好处很实在IDE 可以跳转、断点可以命中、重构可以识别、单元测试可以直接调用函数本身。对一个写过多年 Python 的人来说几乎不需要学习新的心智模型把现有业务函数包一层task就成了。这是我认为 Prefect 最大的设计红利不是教用户怎么适应框架而是让框架来适应你已经会的东西。2. Prefect 的设计突围让 DAG 回归“运行时的直觉”2.1 动态 DAG 不是炫技是真实需求用过静态 DAG 工具的人大多被同一个问题折磨过图必须在解析阶段就被确定下来。Airflow 的调度器会周期性解析 DAG Python 文件把里面定义好的 Task 和依赖关系编译成一张静态图然后再由执行器去跑。听起来没什么问题但真实业务里很多流程只有在运行时才知道要跑哪些任务。举个最典型的例子上游任务扫描数据库发现 80 条新记录需要处理这 80 条记录要并行执行 80 个子任务。在静态 DAG 里你必须在解析阶段就画出这 80 个实例做不到“跑完第一个才知道后面几个”。有人会用动态生成 DAG 文件的方式硬拆但那样代码会变得非常难读而且一不小心就把调度器搞挂。Prefect 的动态 DAG是在 flow 函数真正执行的过程中一边调用任务一边把依赖边画出来。于是你可以放心写循环和分支flow def crawler_flow(): urls load_urls() for url in urls: scrape(url)这里load_urls()返回几条就生成几个scrape任务实例。第一次看到这个设计我只是觉得“方便”后来才意识到它其实是把 DAG 从“预定义的建设蓝图”变成了“运行时留下的执行轨迹”这让写代码的人更容易把注意力放在业务逻辑上而不是放在如何满足调度器上。2.2 Flow 与 Task 的执行模型装饰器如何做到隐形增强Prefect 的核心概念非常简单flow装饰一个函数它就是一个工作流task装饰一个函数它就是工作流里的一个步骤。flow 内部调用 task依赖关系自然生成flow 也可以调用别的 flow相当于把一个大流程拆成若干子流程便于分模块管理。装饰器为什么是好的抽象入口因为它做的是“隐形增强”。被task包裹之后任务函数在调用时会经历状态注册、参数序列化、重试判断、日志记录等一系列过程但对外表现仍然是普通函数该返回什么返回什么该抛异常抛异常。这意味着你完全不需要改写原有业务代码大部分情况下只是“搬家”把函数从一个普通模块搬到 Prefect 的 flow 里。一个值得记住的边界是task 虽然可以脱离 flow 单独被调用但那样它只能记录单次状态享受不到依赖、并发、缓存这些编排层面的能力。所以最佳实践是把所有 task 的调度入口都放在 flow 内。另外task 的粒度也要合适太大不利于复用和观察太小会让流程图变成一堵墙。我一般把“一个完整的数据处理动作”定义成一个 task比如“从某张表读取”“做清洗”“写入目标库”而不是把“加一列”“过滤一行”也拆成 task。2.3 状态机与缓存状态是被持久化的数据不只是 UI 上的小圆点Prefect 里每个 flow run 和 task run 都有一整套状态流转Pending、Running、Completed、Failed、Retrying、Cached、Cancelled、Crashed 等。有人觉得状态只是 UI 上的小圆点我恰恰相反认为状态才是整个框架的地基。因为状态是可持久化、可查询的数据“任务现在怎么样了”就不再依赖人工盯日志而是变成一个普通的数据问题。你可以在代码里根据某个 run 的状态决定下一步动作可以在 UI 里按状态过滤出所有失败任务也可以把这些状态接进自己的监控体系。这一切的前提是框架把状态当作一等公民而不是执行完之后就丢掉的临时信息。缓存机制也是基于状态实现的当系统判断某个任务之前已经用相同的输入参数计算过并且结果还在有效期内它不再重新执行而是直接把新 run 标记为 Cached跳过函数体。这在“多个流程并发去抢同一个结果”的场景特别有价值。比如多个下游都依赖同一个“每月用户画像快照”它们同时触发第一个在算其他的可以直接复用结果省掉的不仅是计算还有排队时间。不过缓存也精准地验证了一句老话越方便的东西越容易踩坑。它能不能命中完全取决于缓存键设计得好不好这部分我在第五章专门展开。3. 动手拆一个真实任务从装饰器到并行执行3.1 最小可用工作流抓数、清洗、入库理论说再多不如跑一个最小流程。假设我们有一个每日订单管道抓取订单数据、清洗、写入数仓用 Prefect 写出来就是这样from prefect import flow, task task def fetch_orders(date: str): # 模拟从上游接口拉数据 print(ffetching orders for {date}) return [{id: 1, amount: 100}, {id: 2, amount: 200}] task def clean_orders(raw_orders: list): # 清洗逻辑去掉金额为负的异常记录 return [o for o in raw_orders if o[amount] 0] task def load_to_warehouse(clean_orders: list): print(floading {len(clean_orders)} records to warehouse) return len(clean_orders) flow def order_pipeline(date: str): raw fetch_orders(date) cleaned clean_orders(raw) load_to_warehouse(cleaned) if __name__ __main__: order_pipeline(2025-01-10)直接运行这段脚本三个 task 会依次执行状态会被记录下来。如果你希望记录能持久化并能在浏览器里看板本地起一个prefect server start再在运行环境里配置好 API 地址即可。只想验证逻辑时直接看打印结果也够了。这个示例虽然短但已经演示了 Prefect 最核心的模型普通函数、函数调用、传参返回。把真实业务里的抓取、清洗、入库都换成真正的实现剩下的骨架完全一致。3.2 依赖是“传参”长出来的不是写进配置的继续看上面的代码你可能已经注意到依赖关系不明显但确实存在。cleaned clean_orders(raw)因为raw来自于上一个 task所以 Prefect 默认把它们连成一条依赖边。这就是我特别欣赏的一点——依赖不是写进一个外部配置文件的而是从代码调用关系里自然长出来的。哪里查依赖你在 flow 里可以用flow.visualize()直接看生成的图这在调试阶段非常好用。不过需要提醒visualize依赖外部渲染库有些环境需要额外安装支持组件跑不出来时别慌从日志顺序也能判断依赖是否正确。还有一类场景是“顺序依赖但不传数据”。比如你希望等三个数据抽取任务全部完成后再发一个通知任务但通知任务本身不需要它们任何返回值。这时可以用wait_for参数指定必须等待的任务这比靠传参硬凑关系更清晰。我建议只在“无数据但有先后约束”的场景用它别拿它替代正常传参否则图会变得难读。3.3 并行执行submit、future 与 Runner 的选择串行能跑通只是第一步真实管道里常要并行。Prefect 里要让 task 并行不需要把函数改成 async只要在 flow 内部使用submit()flow def parallel_pipeline(): a fetch_orders.submit(date2025-01-10) b fetch_refunds.submit(date2025-01-10) combine(a, b) task def combine(orders, refunds): print(forders: {len(orders)}, refunds: {len(refunds)})submit()会立刻返回一个 future两个 task 在后台并发执行当combine接收到 future 作为参数时Prefect 会自动等待它完成并解析出真实结果。注意不要在 flow 里submit()之后什么都不做就结束流程一定要让下游任务或者最终返回依赖这些 future否则可能出现“提交了但没结果”的尴尬。这里的“并行”效果还取决于执行 Runner。默认的 Runner 是线程池IO 密集任务比如网络请求、数据库读取可以明显提速但纯 CPU 计算会被 GIL 拖累线程池几乎不加速。遇到 CPU 密集场景可以给 flow 指定进程池from prefect.task_runners import ProcessPoolTaskRunner flow(task_runnerProcessPoolTaskRunner()) def heavy_pipeline(): ...进程池能让并发性提升不少但代价是参数和返回值需要能序列化比如 DataFrame、大对象要谨慎处理。选哪个 Runner本质上是在“方便”和“性能”之间做权衡建议默认线程池确实跑不满 CPU 再上进程池。3.4 重试、超时、日志与命名把安全阀装在任务上真实世界里网络会抖、数据库会锁、第三方接口会限流所以任务必须自带容错。Prefect 在最常用的位置把重试和超时暴露了出来task( retries3, retry_delay_seconds30, timeout_seconds600, tags[critical] ) def fetch_user_profile(user_id: str): ... return profileretries控制在失败后最多自动重试几次retry_delay_seconds是重试前的等待时间timeout_seconds则给任务设了硬上限超过就放弃并标记失败。这三个参数我几乎所有生产任务都会配齐重试解决偶发失败超时防止任务“挂死但一直不失败”。日志方面直接用 Prefect 的get_run_logger()它会把日志和当前任务/流程 run 绑定UI 里能按 run 检索比在代码里随手print然后再去终端翻日志要强太多。tags 则是给你的任务打标用的比如critical、nightly在 UI 里过滤和做自动化规则时很省事。4. 部署与调度从本地跑通到托管运行4.1 Deployment 的本质同一份代码多种运行形态本地跑通一个 flow 很简单但“每天凌晨自动跑”就是另一回事了。Prefect 引入的 Deployment 概念本质是把“一段 flow 代码”和“一个具体的运行方式”绑定起来给哪个参数、用什么调度、跑在哪个环境。最简单的绑定方式是直接让 flow 常驻并提供调度基于较新的 3.x 写法order_pipeline.serve( nameorders-nightly, cron0 2 * * *, parameters{date: 2025-01-10} )更工程化的链路是prefect deploy在项目根目录初始化配置把 flow 发布到指定工作池由 worker 按调度执行。这条链路的好处是配置可以被版本管理部署流程能自动化和 Code Review适合团队协作。我当时最大的感受是Deployment 把“开发”和“运维”解耦了。写流程的人可以专注业务代码部署的人可以不改代码只调配置同一个 flow 还能同时存在“每天全量跑”和“每分钟增量跑”两种 Deployment互不干扰。这个设计很朴素但非常贴合真实团队的分工。4.2 Work Pool 不是玄学是运行环境的抽象第一次见到 Work Pool 这个词时我也有点懵。其实用劳务市场类比就很清楚了Work Pool 是“需求方”它声明了需要什么样的运行环境比如一个处理任务的进程池、一个 Docker 容器、一个 Kubernetes 集群Worker 是“接单员”它不停轮询 Pool 里有没有待运行的任务有就去跑。本地试验时我们往往用 process 类型的 poolprefect work-pool create local-pool --type process然后让 worker 接单并运行。这个抽象让“调度”和“执行环境”彻底分开生产环境可以是 K8s 集群本地调试可以是进程池同一套代码不用改只是把 Deployment 绑定到不同的 pool 而已。对中小团队我建议先上 process 或 docker 类型把流程跑熟再考虑 K8s。4.3 调度、自动化与 UI被动等待失败的方式可以退休了调度这部分Prefect 同时支持几种风格用cron写传统的定时任务用interval写固定间隔的轮询任务用rrule处理更复杂的日历规则。定时是基础能力真正值得说的是自动化规则。Automations 是 Prefect 的事件响应机制在支持该能力的版本里可以定义“当某个 flow run 失败时往通知渠道发一条消息”“当另一个流程完成后触发下游流程”。这让工作流从“定时到点开跑”进化成了“对状态变化做响应”。UI 也是我长期用下来的加分项。它不是一个只用来“看”的面板还能直接浏览每次 flow run 的输入参数、任务的每次重试记录、日志输出甚至能对比不同调度之间的历史执行情况。我给生产任务的建议是给重要流程打上 tag给关键 task 写足日志每天上班先花三分钟扫一遍失败列表这对稳定性的提升比任何事后救火都有效。5. 生产环境里那些文档不会写的坑5.1 进程一退任务就“蒸发”关于同步等待我见过一个很典型的失误本地调试并行流程时用了submit()但没有让下游或最终返回依赖那些 future脚本跑完直接退出结果控制台看起来什么都没发生去 UI 一查任务状态是 Cancelled 或 Crashed。原因很简单submit()只是把任务提交到后台线程调度主进程一退出后台任务就没了着落。正确做法是在 flow 的最终返回里用future.result()拿到结果或者把 future 传递给下游 task让依赖关系替你做等待。记住一条朴素规则flow 结束之前确保所有提交的任务都明确被等到了。5.2 全局变量在并行 Runner 下的“薛定谔”行为写普通 Python 时大家都习惯用模块级变量但在并行任务里这很危险。线程池下所有任务共享同一个全局变量一个任务改了它另一个任务读到的就是被污染的值进程池下情况更诡异每个进程各有一份副本行为会因为进程隔离而变得难以推测。最稳妥的做法是把任务需要的所有状态都通过参数传入把结果通过返回值传出绝不依赖全局可变对象。这个约束我吃过好几次亏后才真正内化在编排框架里任务函数应该尽量接近“纯函数”至少对外部状态要如此。这样任务才能随时重试、并发、甚至被丢到另一台机器上执行而行为依然一致。5.3 缓存键的边界为什么缓存总是命中不了前面说缓存很好用但它的命中完全依赖cache_key_fn。Prefect 会根据你返回的缓存键去查历史任务记录如果键匹配且没过期就直接复用结果。最常见的错误是缓存键里塞进了时间戳、随机数这些每次都不一样的值# 错误示例每次都不同缓存永远不会命中 def bad_cache_key(context, parameters): return str(datetime.now())正确姿势是让键只和“真正决定计算结果的东西”相关比如业务日期、数据版本号、参数组合。我习惯这样组织from datetime import timedelta def order_cache_key(context, parameters): return forders-v1-{parameters[date]} task(cache_key_fnorder_cache_key, cache_expirationtimedelta(hours6)) def fetch_orders(date: str): ...另一个细节是缓存和重试交互。如果任务在第 3 次重试时成功Prefect 会如何处理之前的失败状态不同版本的策略会有差异我的建议是别把缓存和重试混在一个任务里做实验先用日志确认状态流转再信任缓存结果。5.4 选型对照Airflow、Dagster、Prefect 与裸脚本最后聊一个很多人问的问题到底用 Airflow、Dagster 还是 Prefect或者干脆守住裸脚本我把几个方向的关键差异列成一张表维度AirflowDagsterPrefect裸脚本 cron部署重量重需要 scheduler、web、数据库中依赖元数据服务轻本地可跑生产也灵活最轻DAG 形态静态 DAG 为主以资产和数据血缘为中心动态 DAG运行时生成无 DAG运维成本高适合专岗平台团队中高适合资产治理要求高的团队低到中中小团队友好最低适用规模大规模、长期稳定的批处理平台数据资产复杂、重视血缘快速迭代、脚本密集的中等规模团队极少任务依赖简单扩展能力生态最全强于数据工程治理灵活能与任意 Python 生态融合无我不会替你做决定能说的是我自己的感受如果团队里已经有人专门维护调度平台Airflow 依然是很抗打的选择如果想以数据资产为中心治理Dagster 值得认真评估如果你们是中等规模的数据团队希望用低运维成本把编排做扎实Prefect 几乎是最平滑的上手路径。裸脚本也不是丢人的选项只要任务量真的很少就不要为了用框架而用框架。最后说句实在话选工具不是选一个最能打的而是选一个能让你“早睡早起”的。我自己的判断标准只有三条任务数量超过十个、存在跨天依赖、失败后需要重跑某一段任何一条命中都值得把编排框架纳入技术方案。Prefect 在第一条和第三条上尤其顺手它让流程编排回归了工程师最熟悉的本能用 Python 写逻辑用函数表达流程剩下的交给框架记录与调度。