📚 课程目录第2章 Flow与Task基础

Flow 与 Task 基础：从零搭建你的第一个工作流

说实话，我刚接触 Prefect 的时候，第一反应是：这不就是个带 UI 的装饰器集合吗？但用久了才发现，Flow 和 Task 的设计哲学，才是整个 Prefect 的魂。今天我们就来把这俩东西彻底掰扯清楚。

Flow 的定义与运行

Flow 是什么？说白了，它就是一组 Task 的编排容器。你定义一个 Flow，就像画了一张流程图——谁先跑、谁后跑、谁依赖谁的结果，全由你说了算。

定义一个 Flow 非常简单：

from prefect import flow

@flow(name="我的第一个工作流")
def my_first_flow():
    print("Hello, Prefect!")
    
# 运行它
my_first_flow()

嗯，就这么简单。但注意，@flow 装饰器必须加在顶层函数上。我见过有人把 @flow 加在类方法上，结果跑起来各种报错——Prefect 的调度器只认函数，不认方法。

小技巧： 我个人习惯给每个 Flow 起一个有意义的名字，比如 data_pipeline_v2 而不是 flow_123。这样在 Prefect UI 里排查问题时，一眼就能找到目标。

运行 Flow 的方式有两种：

直接调用：就像调用普通函数一样，适合本地调试。
通过 Prefect API：部署到 Prefect Server 或 Cloud 后，由调度器触发。

直接调用时，Prefect 会在当前进程中创建一个临时运行环境。部署后，它会走完整的调度、重试、通知流程。我建议你在开发阶段先用直接调用，等逻辑稳定了再部署。

Task 装饰器详解

Task 是工作流的最小执行单元。你可以把它理解成一个「带状态追踪的函数」。

from prefect import task

@task(retries=2, retry_delay_seconds=10)
def fetch_data(url: str) -> dict:
    # 模拟网络请求
    return {"status": "ok", "data": url}

这里有几个关键参数，我列个表方便你对照：

参数	作用	我的建议
`retries`	失败后重试次数	网络请求类 Task 至少设 2 次
`retry_delay_seconds`	重试间隔（秒）	别设太短，给下游系统喘息时间
`timeout_seconds`	超时时间	防止 Task 卡死
`tags`	标签，用于过滤和分组	按环境或模块打标签，方便排查
`cache_policy`	缓存策略	幂等 Task 可以开启，省时间

注意： 我曾经把一个 Task 的 retries 设成 10，结果上游数据源挂了，它硬是重试了 10 次才放弃。浪费了整整 5 分钟。现在我的原则是：重试次数不超过 3 次，且必须配合指数退避。

Task 的输入与输出

Task 的输入输出，本质上就是函数的参数和返回值。但 Prefect 在这里做了两件聪明的事：

自动类型推断：你不需要手动声明参数类型，Prefect 会通过类型注解自动处理。
结果持久化：Task 的返回值会被 Prefect 自动存储，方便后续 Task 引用。

@task
def add(x: int, y: int) -> int:
    return x + y

@task
def multiply(z: int, factor: int) -> int:
    return z * factor

@flow
def compute_flow():
    result1 = add(3, 5)        # result1 是 int
    result2 = multiply(result1, 2)  # 直接传 result1
    print(result2)  # 输出 16

你想想看，这里最妙的是什么？result1 并不是一个普通的 int，而是一个 Prefect 的 Future 对象。当你把它传给 multiply 时，Prefect 会自动解析它的实际值。这种「延迟求值」的机制，让依赖关系的表达变得极其自然。

  核心概念： Prefect 的 Task 返回值是「惰性的」。它不会立即计算，而是在 Flow 运行时才解析。这保证了依赖关系的正确性，也让你能写出更声明式的代码。

Task 的依赖关系

依赖关系是工作流的骨架。在 Prefect 里，依赖关系通过两种方式表达：

隐式依赖：一个 Task 的输入来自另一个 Task 的输出。
显式依赖：使用 wait_for 参数强制指定前置 Task。

@task
def step_a():
    return "A done"

@task
def step_b():
    return "B done"

@task
def step_c(a_result: str, b_result: str):
    return f"{a_result} + {b_result}"

@flow
def dependency_flow():
    a = step_a()
    b = step_b()
    c = step_c(a, b)  # 隐式依赖：c 依赖 a 和 b

这里 step_a 和 step_b 是并行执行的。只有两者都完成后，step_c 才会启动。Prefect 的调度器会自动分析这些依赖关系，构建出一个 DAG（有向无环图）。

那什么时候用显式依赖？举个例子：

@task
def send_notification():
    print("通知已发送")

@flow
def explicit_flow():
    a = step_a()
    b = step_b(wait_for=[a])  # 显式等待 a 完成
    send_notification(wait_for=[a, b])  # 等待 a 和 b 都完成

send_notification 不需要任何输入，但它必须在 a 和 b 之后执行。这时候 wait_for 就派上用场了。

避坑指南： 我曾经在一个复杂的 ETL 流程里，忘了加 wait_for，结果通知 Task 在数据还没写完时就触发了。下游系统收到空文件，报警响了一整夜。从那以后，只要 Task 有隐含的顺序要求，我一定加显式依赖。

Flow 的嵌套

Flow 可以嵌套，这是 Prefect 最强大的特性之一。你可以把一组相关的 Task 封装成一个子 Flow，然后在父 Flow 中调用它。

@flow(name="数据清洗")
def clean_flow(raw_data: list) -> list:
    # 清洗逻辑
    return [d for d in raw_data if d.get("valid")]

@flow(name="数据转换")
def transform_flow(clean_data: list) -> list:
    # 转换逻辑
    return [{"id": d["id"], "value": d["value"] * 2} for d in clean_data]

@flow(name="主流程")
def main_flow():
    raw = [{"id": 1, "valid": True, "value": 10}, {"id": 2, "valid": False, "value": 20}]
    clean = clean_flow(raw)
    result = transform_flow(clean)
    print(result)

嵌套 Flow 的好处是什么？

模块化：每个子 Flow 可以独立测试、独立部署。
可观测性：在 Prefect UI 里，你可以展开每个子 Flow，看到它内部的 Task 执行情况。
复用性：同一个子 Flow 可以在多个父 Flow 中调用。

我个人习惯把「数据清洗」「特征工程」「模型训练」这些逻辑上独立的阶段，都封装成子 Flow。这样主流程看起来就像一份清晰的目录，而不是一团乱麻。

经验之谈： 嵌套深度不要超过 3 层。我见过有人嵌套了 5 层，结果排查问题时，点开一层又一层，像剥洋葱一样。保持扁平化，能让你和你的同事都少掉头发。

本章知识体系

下面这张图，把 Flow 和 Task 的核心关系画出来了：

从这张图你能看到，Flow 是树状结构，Task 是叶子节点。父 Flow 负责编排，子 Flow 负责模块化，Task 负责具体执行。这种分层设计，让复杂的工作流变得清晰可控。