📚 课程目录第4章 Flow运行配置

4、Flow运行配置：参数传递、运行器、并发控制与标签分组

好，咱们进入第四章。这一章我打算把 Flow 的运行配置讲透。说白了，就是当你写好一个 Flow 之后，怎么让它跑起来，怎么传参数，怎么控制它是一次跑一个任务还是一窝蜂全上，以及怎么给 Flow 打标签做分组。

这些内容在实际项目中特别重要。我记得刚用 Prefect 那会儿，就因为没搞清楚参数传递的坑，导致生产环境跑出来的数据全是错的……嗯，从那以后我就老老实实把这块啃透了。

4.1 Flow 参数传递：不只是传个值那么简单

先看最基础的部分。你定义一个 Flow 函数，参数怎么传？

from prefect import flow

@flow
def my_flow(name: str, age: int = 18):
    print(f"Hello {name}, age {age}")

# 直接调用
my_flow("Alice", 25)
# 或者用关键字
my_flow(name="Bob", age=30)

看起来跟普通 Python 函数没区别对吧？但这里有个关键点：Prefect 会序列化这些参数。你传进去的每个参数，都会被 Prefect 记录下来，存在它的后台数据库里。这意味着什么？意味着你可以回溯每次运行的参数值，方便调试。

我的习惯：所有关键参数都显式写在 Flow 函数的签名里，不要用 **kwargs 一把抓。这样 Prefect 才能完整记录参数历史，排查问题的时候一目了然。

4.2 参数类型与验证

Prefect 支持的类型很全：str、int、float、bool、list、dict、datetime，甚至 Pydantic 模型。我个人强烈建议用 Pydantic 做复杂参数校验。

from pydantic import BaseModel
from prefect import flow

class Config(BaseModel):
    batch_size: int = 32
    learning_rate: float = 0.001
    use_gpu: bool = True

@flow
def training_flow(cfg: Config):
    print(f"Batch size: {cfg.batch_size}")
    # 你的训练逻辑...

为什么推荐这个？因为 Pydantic 会在参数传入时自动做类型校验。我曾经在生产环境遇到过传错参数类型导致整个 pipeline 崩溃的情况，用了 Pydantic 之后这类问题基本绝迹了。

4.3 FlowRunner：控制 Flow 怎么跑

FlowRunner 是 Prefect 2.x 里控制 Flow 执行的核心组件。你写 @flow 装饰器的时候，其实就是在配置一个 FlowRunner。

参数	作用	我的常用值
`retries`	失败重试次数	2 或 3
`retry_delay_seconds`	重试间隔	30 或 60
`timeout_seconds`	超时时间	根据任务定，一般 3600
`log_prints`	是否捕获 print	True
`validate_parameters`	是否校验参数	True

@flow(
    retries=2,
    retry_delay_seconds=30,
    timeout_seconds=3600,
    log_prints=True,
    validate_parameters=True
)
def robust_flow(data_path: str):
    print(f"Processing {data_path}")
    # 你的业务逻辑

注意：timeout_seconds 是 Flow 级别的超时，不是单个 task 的超时。如果你 Flow 里有多个 task，总执行时间超过这个值就会被强制终止。我曾经因为没设这个参数，一个死循环的 Flow 跑了整整两天才发现……

4.4 并发与顺序执行控制

这是很多初学者容易搞混的地方。Prefect 里控制并发有两种维度：task 级别和Flow 级别。

4.4.1 Task 级别的并发

默认情况下，同一个 Flow 里的 task 是按依赖关系顺序执行的。但如果你想让某些不相关的 task 并行跑，可以用 task_runner 参数。

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@flow(task_runner=ConcurrentTaskRunner)
def parallel_flow():
    # 这三个 task 没有依赖关系，会并行执行
    result_a = task_a.submit()
    result_b = task_b.submit()
    result_c = task_c.submit()

这里有个细节：submit() 返回的是一个 PrefectFuture 对象，你需要调用 .result() 才能拿到实际值。我刚开始用的时候经常忘记调 .result()，结果拿到的全是 future 对象，调试了半天……

4.4.2 Flow 级别的并发控制

如果你同时启动多个 Flow 实例，可以用 concurrency_limit 来控制同时运行的 Flow 数量。

from prefect.concurrency.sync import concurrency

@flow
def limited_flow():
    with concurrency("my_limit", max_out=3):
        # 最多同时 3 个 Flow 实例能进入这个代码块
        do_something()

这个功能在资源有限的环境下特别有用。比如你只有 4 个 GPU，那并发跑 4 个训练任务就够了，再多反而互相拖慢。

4.5 Flow 的标签与分组

当你的项目越来越大，Flow 越来越多，怎么管理它们？标签和分组就是答案。

4.5.1 标签（Tags）

标签是给 Flow 打上的元数据，可以用来过滤、搜索、权限控制。

@flow(tags=["production", "gpu", "training"])
def production_training_flow():
    pass

@flow(tags=["staging", "cpu", "testing"])
def staging_test_flow():
    pass

在 Prefect UI 里，你可以按标签筛选 Flow 运行记录。我个人习惯给每个环境（dev/staging/prod）和每个业务模块都打上标签，这样排查问题的时候能快速定位。

4.5.2 分组（Groups）

分组比标签更结构化。你可以把相关的 Flow 放在同一个项目或部署组里。

from prefect import flow

@flow
def data_ingestion():
    pass

@flow
def data_cleaning():
    pass

# 这两个 Flow 可以通过部署时的 project 参数分组
# prefect deployment build data_ingestion:main -n "ingestion" -p "data_pipeline"
# prefect deployment build data_cleaning:main -n "cleaning" -p "data_pipeline"

  实战建议：标签和分组配合使用效果最好。标签描述「是什么」（比如：gpu、critical），分组描述「属于哪个系统」（比如：data_pipeline、ml_training）。这样你既可以从系统维度看，也可以从属性维度看。

4.6 本章知识体系图

下面这张图把本章的核心知识点串起来了，方便你对照理解。

4.7 避坑指南

最后分享几个我踩过的坑，希望能帮你少走弯路。

参数默认值陷阱：不要在 Flow 函数参数里用可变对象做默认值（比如 def f(x=[])），Prefect 序列化时会出问题。用 None 然后在函数内部初始化。
并发数不是越大越好：我曾经把并发数设成 100，结果数据库连接池直接被打满，所有任务都挂了。建议根据下游系统的承载能力来设置。
标签命名规范：团队协作时，标签命名一定要统一。我们团队吃过亏，有人用 "prod"，有人用 "production"，结果筛选的时候漏掉了一半的 Flow。
超时时间要留余量：别把 timeout_seconds 设得太紧。数据量波动、网络抖动都可能导致执行时间变长，留 20%-30% 的余量比较安全。