Prefect

概述

Prefect 是一个用于构建 Python 数据管道的工作流编排框架。它是将脚本提升为交互式工作流应用的最简单方式，可帮助用户构建具备弹性、动态特性的工作流，能够响应周围环境变化并从意外情况中恢复。

通过少量装饰器，Prefect 可为代码增强多种功能，如自动重试、分布式执行、调度、缓存等。

工作流活动可通过自托管的 Prefect server 实例或托管的 Prefect Cloud 仪表板进行跟踪和监控。

核心特性

• 弹性工作流：自动处理故障并恢复，适应动态变化
• 丰富功能集：支持自动重试、分布式执行、定时调度、结果缓存等
• 简单集成：通过 Python 装饰器轻松增强现有代码
• 可视化监控：提供自托管服务器或云端仪表板，实时跟踪工作流状态
• 灵活部署：支持手动触发、定时调度及事件驱动执行（需 Prefect Cloud）

快速开始

环境要求

Prefect 要求 Python 3.9 或更高版本。

安装

如需安装或升级到最新版本的 Prefect，运行以下命令：

pip install -U prefect

示例工作流

创建并运行一个使用 Prefect flow 和 task 装饰器的 Python 文件，编排和观察工作流。以下示例脚本用于获取 GitHub 仓库的星标数量：

from prefect import flow, task
from typing import List
import httpx


@task(log_prints=True)
def get_stars(repo: str):
    url = f"[***]{repo}"
    count = httpx.get(url).json()["stargazers_count"]
    print(f"{repo} has {count} stars!")


@flow(name="GitHub Stars")
def github_stars(repos: List[str]):
    for repo in repos:
        get_stars(repo)


# 运行工作流
if __name__ == "__main__":
    github_stars(["PrefectHQ/Prefect"])

运行与监控

启动 Prefect UI 查看工作流运行情况：

prefect server start

部署与调度

如需按计划运行工作流，可将其转换为部署并设置每分钟运行一次。修改脚本最后一行为：

if __name__ == "__main__":
    github_stars.serve(name="first-deployment", cron="* * * * *")

此时本地服务器将监听计划部署，可通过 UI 或 CLI 手动触发工作流。若使用 Prefect Cloud，还可配置 事件响应 触发部署。

Docker 部署方案示例

Dockerfile

FROM python:3.9-slim

WORKDIR /app

# 安装依赖
RUN pip install -U prefect httpx

# 复制工作流代码（假设代码文件名为 github_stars.py）
COPY github_stars.py .

# 运行部署
CMD ["python", "github_stars.py"]

构建与运行

1. 构建镜像：

docker build -t prefect-github-stars .

2. 运行容器（映射 4200 端口以访问 Prefect UI）：

docker run -d -p 4200:4200 --name prefect-worker prefect-github-stars

Docker Compose 示例

version: '3.8'

services:
  prefect-worker:
    build: .
    ports:
      - "4200:4200"
    volumes:
      - ./github_stars.py:/app/github_stars.py  # 挂载代码文件，方便修改
    restart: unless-stopped

社区资源

• GitHub 仓库
• Slack 社区