1、Prefect入门：工作流引擎概述、Prefect发展史、Prefect核心概念（Flow、Task、Deployment）、安装与Hello World

1.1 工作流引擎到底是个啥？

先聊聊工作流引擎。说白了，它就是帮你把一堆任务串起来、管起来的工具。你想想看，数据工程里经常有这种场景：先爬数据，再清洗，然后做特征工程，最后训练模型。每一步都可能失败，每一步都可能跑很久。

我刚开始做数据管道的时候，用的是Shell脚本加crontab。结果呢？某天凌晨三点，中间步骤挂了，后面的任务全白跑。第二天上班一看日志，头都大了。后来我意识到，我们需要一个能自动重试、能监控状态、能调度依赖的东西——这就是工作流引擎。

Prefect就是干这个的。它帮你定义任务之间的依赖关系，自动处理失败重试，还能实时告诉你每个任务跑到哪一步了。嗯，这里要注意，它不是唯一的选择，但它在易用性和功能之间找到了一个很好的平衡点。

1.2 Prefect的发展史：从学术项目到行业标配

Prefect的故事挺有意思的。它最早是2018年从一家叫Prefect Technologies的公司孵化出来的。创始人是Jeremiah Lowin，他之前在NASA做数据工程，深知现有工作流工具的痛点。

我记得Prefect 1.0刚出来的时候，最大的卖点是“动态工作流”——你可以在运行时动态生成任务，而不是像Airflow那样必须提前画好DAG。这个特性在当时很前卫。

到了2022年，Prefect 2.0发布，是一次彻底的重写。API更简洁，去掉了大量样板代码。我个人习惯用2.0之后的版本，因为写起来真的舒服很多。

现在Prefect已经成了数据工程圈的主流选择之一。社区活跃，文档完善，企业版也做得不错。如果你关注过2023-2024年的技术趋势，会发现很多公司都在从Airflow迁移到Prefect。

版本	发布时间	核心变化
Prefect 0.x	2018	初始版本，动态DAG概念
Prefect 1.0	2020	稳定版，Orion引擎
Prefect 2.0	2022	完全重写，简化API
Prefect 3.0	2024	事件驱动，增强监控

1.3 三个核心概念：Flow、Task、Deployment

Prefect的核心概念其实就三个：Flow、Task、Deployment。搞懂这三个，你就入门了。

1.3.1 Task：最小的执行单元

Task就是你的一个具体操作。比如“下载文件”、“清洗数据”、“发送邮件”。在Prefect里，你只需要在函数上加一个@task装饰器，它就变成了一个可被管理的任务。

我曾经在一个项目中，把一百多个函数都加上了@task，结果发现有些任务根本不需要重试或监控。后来我学乖了，只对关键步骤加装饰器。你想想看，不是所有函数都值得变成Task的。

from prefect import task

@task(retries=2, retry_delay_seconds=10)
def fetch_data(url: str) -> dict:
    # 这里写你的数据抓取逻辑
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

1.3.2 Flow：任务的编排者

Flow把多个Task串起来，定义它们的执行顺序和依赖关系。你可以在Flow里写条件判断、循环、甚至动态生成任务。

我个人习惯把Flow看作一个“剧本”，Task是“演员”，Flow决定了谁先上场、谁后上场、演砸了怎么办。

from prefect import flow

@flow
def my_data_pipeline():
    raw_data = fetch_data("https://api.example.com/data")
    cleaned_data = clean_data(raw_data)
    result = analyze_data(cleaned_data)
    return result

注意看，这里没有显式声明依赖。Prefect会自动分析函数调用关系，构建出执行图。这就是它比Airflow方便的地方——你不用手动写>>操作符。

1.3.3 Deployment：让Flow跑起来

Deployment是Prefect 2.0引入的概念。它把Flow、调度规则、运行环境打包成一个可部署的单元。你可以把它理解成“生产环境配置”。

举个例子，你开发完一个Flow，想在每天凌晨3点跑一次，并且用Docker容器运行。这些配置就写在Deployment里。

from prefect.deployments import Deployment
from prefect.filesystems import GitHub

deployment = Deployment.build_from_flow(
    flow=my_data_pipeline,
    name="production",
    schedule={"cron": "0 3 * * *"},
    storage=GitHub(repository="my-org/my-repo"),
    work_queue_name="default"
)

1.4 安装与Hello World

好了，理论说完了，咱们动手试试。安装Prefect非常简单，一行命令搞定。

pip install prefect

我建议你在虚拟环境里装，别直接装到系统Python里。我曾经有一次把依赖搞乱了，花了半天才恢复。

装完之后，验证一下：

prefect version

如果看到版本号，说明安装成功。接下来写你的第一个Prefect程序。

1.4.1 Hello World：最简单的Flow

from prefect import flow, task

@task
def say_hello(name: str) -> str:
    message = f"Hello, {name}!"
    print(message)
    return message

@flow
def hello_flow(names: list[str]):
    for name in names:
        say_hello(name)

if __name__ == "__main__":
    hello_flow(["Prefect", "World", "Data Engineer"])

运行这个脚本，你会看到Prefect自动记录每次调用，生成运行ID，还能在终端看到实时日志。嗯，这里要注意，第一次运行可能会慢一点，因为Prefect要初始化一些元数据。

1.4.2 启动Prefect Server（可选）

如果你想看到漂亮的UI界面，可以启动Prefect Server：

prefect server start

然后在浏览器打开 http://localhost:4200。你会看到刚才运行的Flow记录，包括每个Task的状态、耗时、日志。我个人习惯开发时开着Server，方便调试。

1.5 知识体系总览

下面这张图帮你理清本章的知识结构。我画的时候特意把核心概念放在中间，外围是安装和Hello World的步骤。

1.6 避坑指南