prefect

Prefect 数据工作流编排

补充说明

Prefect 用 Python 装饰器定义 flow 和 task，强调动态工作流、重试和可观测性。Prefect 3.x 使用 work pool 和 worker 执行部署，适合数据管道和轻量级编排。

常用示例

启动本地服务：

prefect server start
prefect config set PREFECT_API_URL=http://127.0.0.1:4200/api

定义并运行 flow：

from prefect import flow, task

@task(retries=3, retry_delay_seconds=10)
def extract():
    return [1, 2, 3]

@flow
def pipeline():
    data = extract()
    print(sum(data))

if __name__ == "__main__":
    pipeline()

创建 work pool 和部署：

prefect work-pool create local --type process
prefect deploy ./pipeline.py:pipeline --name daily --pool local
prefect worker start --pool local

触发和查看：

prefect deployment run 'pipeline/daily'
prefect flow-run ls
prefect flow-run inspect <FLOW_RUN_ID>

设置定时调度：

prefect deployment schedule create 'pipeline/daily' \
  --cron '0 2 * * *' --timezone Asia/Shanghai

易错点与排错

• Worker 没有启动或 work pool 名称不匹配，部署一直处于 Late。
• PREFECT_API_URL 指向错误地址，CLI 和 worker 连不上服务端。
• Task 里做不可重入的副作用，重试导致重复写入。
• 部署代码没有同步到 worker 环境，运行时找不到模块。
• 使用 SQLite 作为生产元数据库，并发高时锁冲突。
• 调度时区使用默认 UTC，和业务预期差 8 小时。
• 日志没有接入外部存储，容器删除后无法追溯。
• 依赖版本未锁定，worker 重建后行为变化。