Flow 与 Task 基础：装饰器与依赖关系

这一章，我们正式进入 Prefect 的核心地带。说白了，就是搞清楚两个最基础的装饰器——@flow 和 @task。它们就像乐高积木里的基础块，所有复杂的工作流都靠它们搭起来。

我个人习惯把 Flow 想象成一个「项目经理」，Task 则是「具体干活的员工」。项目经理不干活，但他知道谁先干、谁后干、结果给谁。嗯，这个比喻虽然糙，但道理不糙。

@flow 装饰器：工作流的入口

@flow 是 Prefect 里最顶层的装饰器。你写的任何一个编排逻辑，都得用它包一层。为什么？因为 Prefect 要靠它来识别「这是一个工作流」，然后才能做状态跟踪、重试、调度这些事。

from prefect import flow

@flow(name="我的第一个工作流")
def my_first_flow():
    print("Hello, Prefect!")
    return "完成"

if __name__ == "__main__":
    my_first_flow()

你看，就这么简单。但这里有个坑——name 参数。我在项目中遇到过好几次，忘了给 Flow 起名字，结果在 Prefect Server 的 UI 里看到一堆 unnamed-flow-xxx，根本分不清谁是谁。所以我的建议是：每个 Flow 都显式给个名字，哪怕是个简单的 demo。

@flow 还支持不少参数，我挑几个常用的列出来：

参数	类型	说明
name	str	Flow 的名称，必填建议
description	str	描述这个 Flow 是干什么的
retries	int	整个 Flow 失败后的重试次数
retry_delay_seconds	int/float	重试间隔，单位秒
log_prints	bool	是否把 print 输出记到日志里

@task 装饰器：真正干活的单元

Task 是工作流里的最小执行单元。一个 Flow 里可以有一个 Task，也可以有成百上千个。每个 Task 负责一件具体的事——比如拉数据、做计算、发通知。

from prefect import flow, task

@task(name="数据清洗")
def clean_data(raw_data: list) -> list:
    # 模拟清洗逻辑
    cleaned = [x for x in raw_data if x is not None]
    return cleaned

@task(name="数据聚合")
def aggregate_data(cleaned_data: list) -> dict:
    # 模拟聚合逻辑
    return {"count": len(cleaned_data), "items": cleaned_data}

@flow(name="数据处理流水线")
def data_pipeline():
    raw = [1, 2, None, 3, None, 4]
    cleaned = clean_data(raw)
    result = aggregate_data(cleaned)
    print(result)

if __name__ == "__main__":
    data_pipeline()

你想想看，这里有个关键点：clean_data 和 aggregate_data 虽然是普通的 Python 函数，但加上 @task 后，Prefect 就能追踪它们的运行状态、耗时、返回值。这就是装饰器的魔力。

Task 的参数和 Flow 类似，但多了几个跟执行相关的：

参数	说明
name	Task 名称，建议显式指定
retries	这个 Task 失败后的重试次数
retry_delay_seconds	重试间隔
timeout_seconds	超时时间，超时自动标记失败
cache_key_fn	缓存策略函数，避免重复计算

任务依赖关系：谁先谁后

工作流编排的核心，就是定义任务之间的依赖关系。Prefect 的依赖关系是隐式的——你不需要显式写 .set_downstream() 之类的方法，只要把 Task 的返回值传给另一个 Task，Prefect 就自动知道谁先谁后。

@task
def step_one():
    return "第一步结果"

@task
def step_two(input_data: str):
    return f"第二步处理: {input_data}"

@task
def step_three(input_data: str):
    return f"第三步处理: {input_data}"

@flow
def demo_dependencies():
    a = step_one()
    # step_two 和 step_three 都依赖 step_one 的结果
    b = step_two(a)
    c = step_three(a)
    # 这里 b 和 c 是并行执行的
    return b, c

为什么会这样？因为 Prefect 在运行 Flow 时，会先构建一个 DAG（有向无环图）。它分析每个 Task 的输入参数，如果参数是另一个 Task 的返回值，那就建立一条依赖边。没有依赖关系的 Task，默认并行执行。

我遇到过不少新手问：「怎么让 Task B 在 Task A 之后执行，但又不传参数？」我的回答是：那就传个没用的参数。比如让 Task B 接收 Task A 返回的一个简单标记，纯粹为了建立依赖。

返回值传递：数据怎么流动

Task 之间的数据传递，说白了就是函数调用。Prefect 会把 Task 的返回值包装成一个 PrefectFuture 对象，但你在代码里用的时候，它表现得就像普通值一样。

@task
def fetch_data() -> dict:
    return {"user": "张三", "score": 95}

@task
def process_data(data: dict) -> str:
    return f"用户 {data['user']} 的分数是 {data['score']}"

@flow
def data_flow():
    raw = fetch_data()
    # raw 在这里是 PrefectFuture，但你可以直接当 dict 用
    result = process_data(raw)
    print(result)

这里有个细节：raw 的类型其实是 PrefectFuture[dict]，不是 dict。但 Prefect 在运行时会自动解包，所以你写代码时感觉就像在操作普通数据。嗯，这个设计挺贴心的，省了不少心智负担。

不过要注意一点——Task 的返回值必须是可序列化的。Prefect 需要把结果存到数据库里，如果你返回一个数据库连接对象或者一个打开的文件句柄，那就会报错。我在项目中吃过这个亏，返回了一个 SQLAlchemy Session，结果 Prefect 序列化失败，整个 Flow 崩了。

知识体系总览

下面这张图把本章的核心概念串起来了。你可以看到 Flow 在最外层，里面包含多个 Task，Task 之间通过返回值建立依赖，最终形成一个有向无环图。

避坑指南

最后，我把自己踩过的坑整理一下，希望能帮你少走弯路：

• 不要在 Task 内部调用另一个 Task 函数——应该把 Task 的调用放在 Flow 里，否则 Prefect 无法追踪依赖关系。
• Task 返回值不要太大——Prefect 会把返回值存到后端存储里，几 MB 的数据没问题，但几百 MB 的数据会拖垮系统。
• 注意 Task 的幂等性——如果 Task 有重试机制，要保证它多次执行的结果是一样的，否则数据会乱。
• Flow 里不要写耗时操作——Flow 本身应该只负责编排，真正的计算逻辑放到 Task 里。

好了，这一章的内容就这些。记住：Flow 是骨架，Task 是血肉，返回值是血管。把这三样东西搞明白，你就掌握了 Prefect 工作流编排的七成功力。