Python工作流引擎怎么选

本文深入剖析四大主流Python工作流引擎（Airflow、Prefect、Luigi及自研方案）的真实适用边界与高频踩坑点：Airflow并非万能，面对简单定时任务或小团队时反而笨重难维；Prefect的“语义陷阱”常让开发者误以为代码执行即被调度，实则因装饰器缺失或副作用导致任务悄然失效；Luigi对依赖声明极度严苛，requires()返回非Task实例将引发静默失败；而自研调度器仅在已有执行框架需轻量集成或秒级高频调度等极少数场景下才具合理性——帮你避开选型幻觉，用对工具，而非用尽工具。

什么时候该避开 airflow？

如果你的任务调度周期固定、依赖少、不涉及跨系统数据搬运，或者团队里没人愿意碰 DAG 定义和 airflow.cfg 里的 executor 配置，airflow 就是杀鸡用牛刀。它启动慢、Web UI 依赖完整服务栈、单机模式下连 LocalExecutor 都可能因并发数错配导致任务卡住。

• 典型踩坑：用 SequentialExecutor 跑多任务，结果所有任务串行排队，以为“调度器坏了”
• 真实场景：每天凌晨跑一次 ETL，源是 MySQL，目标是 CSV 归档 —— schedule.every().day.at("02:00").do(...) + APScheduler 更轻
• 兼容性注意：airflow 2.0+ 强制 Python 3.8+，且插件生态对 PyPy/conda 环境支持弱

prefect 的 Task 和 Flow 为什么容易写错？

不是语法错，是语义错：把本该是纯函数逻辑的 Task 写成带状态副作用的操作（比如在 Task 里直接改全局变量），或在 Flow 顶层调用未装饰的函数，导致调度时根本不会被追踪。

• 常见错误现象：Flow 运行后日志显示 “No tasks submitted”，实际代码已执行 —— 因为没用 @task 装饰器
• 参数差异：@task(retries=3, retry_delay_seconds=10) 是任务级重试；@flow(retry_policy=RetryPolicy(max_retries=1)) 是整个 Flow 失败后重跑，二者不叠加
• 性能影响：本地运行时默认用 ConcurrentTaskRunner，但若任务含大量 I/O（如频繁读文件），反而不如 SequentialTaskRunner 稳定

用 luigi 写依赖链，requires() 返回什么才安全？

必须返回 Task 实例，不能返回列表推导式结果、也不能返回未实例化的类（比如写成 requires(self): return [MyTask]）。否则调度器无法解析依赖图，会静默跳过下游任务。

• 典型错误：requires() 里用 os.path.exists() 做前置判断，结果路径不存在时返回空列表 —— luigi 当作“无依赖”，直接执行当前任务，而非报错阻断
• 使用场景：ETL 流程中“清洗 → 校验 → 加载”，每个环节输出一个 Target（如 LocalTarget("data/cleaned.parquet")），requires() 就得对应返回生成该 Target 的 Task 实例
• 兼容性注意：luigi 不处理任务失败后的自动回滚，得自己在 run() 里写 try/except 清理中间文件

自研调度器真有必要吗？

只在两种情况下值得考虑：一是已有成熟任务执行框架（比如公司内部的 Python 批处理服务），只需加一层 cron+状态轮询；二是任务粒度极细（秒级触发、单次执行airflow 或 prefect 的调度开销（DB 查询、序列化、心跳）已占耗时 30% 以上。

• 容易被忽略的复杂点：分布式锁。多个 worker 同时抢同一个任务时，靠文件或 DB 行锁很容易漏掉竞态，redis.lock() 是目前最稳的轻量方案
• 别低估的维护成本：一旦加了重试、超时、优先级、资源限制，代码量会快速逼近 celery 的简化版，而你还没法复用它的监控和 CLI 工具
• 真实建议：先用 APScheduler + sqlite 持久化撑三个月，等任务数破 50、失败率超 5%，再评估迁移

本篇关于《Python工作流引擎怎么选》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！