📚 课程目录第4章 Flow配置

Flow 配置：给工作流一个身份

Flow 配置这件事，说白了就是给你的工作流一个「身份证」。我刚开始用 Prefect 时，觉得能跑就行，名字随便起、版本懒得写。结果项目一多，十几个 flow 堆在一起，根本分不清哪个是哪个。嗯，后来我花了一整天重构配置，从那以后，每个 flow 我都认真对待。

这一章，我们聊聊 flow 的四个核心配置项：名称与描述、版本管理、标签与分组、调度策略。它们看似简单，但用好了，能让你的工作流编排井井有条。

4.1 Flow 名称与描述：第一印象很重要

每个 flow 都应该有一个清晰的名字。我个人习惯用「动词+名词」的格式，比如 process_payment、sync_user_data。别用 flow_1、test_flow_final 这种名字，你想想看，三个月后你自己都看不懂。

命名规范建议：

使用小写字母 + 下划线（snake_case）
前缀表示业务域，如 payment_、analytics_
动词开头，明确动作
长度控制在 3-5 个单词以内

描述字段很多人会忽略，但我建议你写。哪怕只有一句话，也能帮队友（包括未来的你）快速理解这个 flow 是干什么的。

from prefect import flow

@flow(
    name="payment.process_refund",
    description="处理用户退款请求，包含风控校验和资金回滚"
)
def process_refund(order_id: str, amount: float):
    """处理退款流程"""
    pass

我的小技巧：描述里可以写清楚「输入是什么」「输出是什么」「失败时怎么办」。比如「处理退款请求，输入 order_id，输出退款结果，失败时发送告警」。这样别人看描述就知道怎么用。

4.2 Flow 版本管理：别让历史成为负担

版本管理这件事，我曾经吃过亏。有一次线上 flow 出了 bug，我想回滚到上一个版本，结果发现所有 flow 都叫 v1，根本不知道哪个是哪个。后来我强制团队使用语义化版本号。

Prefect 支持在 @flow 装饰器中指定版本号。我个人习惯用 major.minor.patch 格式：

major：重大重构，接口不兼容
minor：新增功能，向后兼容
patch：bug 修复，小改动

@flow(
    name="data.sync_user",
    version="2.1.0",
    description="同步用户数据到数仓，v2 版本新增增量同步支持"
)
def sync_user():
    pass

注意：Prefect 的版本号只是元数据，不会自动帮你做版本回滚。它更像一个「标签」，方便你在 UI 上区分不同版本的 flow。真正的版本控制，还是要靠 Git。

我建议你在每次修改 flow 逻辑时，同步更新版本号。可以写一个简单的脚本，在 CI/CD 流程中自动递增版本号。这样就不会出现「改了代码忘了改版本」的尴尬。

4.3 Flow 标签与分组：给工作流打上标签

标签是 Prefect 里非常实用的功能。你可以把标签理解为「分类标签」或「环境标识」。比如：

production、staging、development 表示环境
critical、batch、realtime 表示优先级或类型
team-a、team-b 表示归属团队

@flow(
    name="etl.daily_report",
    tags=["production", "batch", "team-analytics"]
)
def daily_report():
    pass

标签有什么用？你想想看，当你有几十个 flow 时，在 UI 上按标签筛选，瞬间就能找到你要的 flow。我曾经在排查问题时，通过标签快速过滤出所有 critical 级别的 flow，优先处理它们。

标签最佳实践：

每个 flow 至少打 2-3 个标签
标签命名统一，不要出现 prod 和 production 混用
用标签区分「环境」和「用途」两个维度
避免标签过多，5 个以内比较合适

4.4 Flow 调度策略：让工作流按时执行

调度是 flow 的核心能力之一。Prefect 支持多种调度方式，我挑几个常用的说说。

调度方式	适用场景	示例
Cron 表达式	固定时间间隔，如每天凌晨 2 点	`0 2 * * *`
Interval	相对时间间隔，如每 30 分钟	`minutes=30`
Rrule	复杂重复规则，如每月第一个周一	`rrule=freq=MONTHLY;byweekday=MO`
事件驱动	外部触发，如文件到达、Webhook	配合 `@trigger` 使用

我个人最常用的是 Cron 表达式。它灵活、通用，而且大多数工程师都熟悉。但要注意时区问题——我曾经踩过这个坑，Cron 默认是 UTC 时间，如果你在中国，记得加上时区参数。

from datetime import timedelta
from prefect import flow
from prefect.schedules import CronSchedule, IntervalSchedule

# Cron 调度：每天北京时间 8:00 执行
@flow(
    name="daily_report",
    schedule=CronSchedule(cron="0 0 * * *", timezone="Asia/Shanghai")
)
def daily_report():
    pass

# Interval 调度：每 30 分钟执行一次
@flow(
    name="health_check",
    schedule=IntervalSchedule(interval=timedelta(minutes=30))
)
def health_check():
    pass

避坑指南：我曾经在生产环境设置了一个 Cron 调度，结果 flow 在凌晨 3 点疯狂重试。排查了半天，发现是 Cron 表达式写错了——* * * * * 表示每分钟执行一次，而不是每天。嗯，从那以后我每次写 Cron 都会用在线工具验证一遍。

4.5 知识体系总览

下面这张图总结了 flow 配置的四个核心维度，以及它们之间的关系。你可以把它当作一个快速参考。

4.6 综合示例：一个完整的 Flow 配置

最后，给你看一个我实际项目中的 flow 配置。它包含了我们上面讲的所有要素：

from datetime import timedelta
from prefect import flow
from prefect.schedules import CronSchedule

@flow(
    name="analytics.daily_kpi",
    description="每日 KPI 计算：从数仓读取原始数据，计算核心指标，写入报表库。失败时发送企业告警。",
    version="3.2.0",
    tags=["production", "batch", "team-analytics", "critical"],
    schedule=CronSchedule(cron="30 1 * * *", timezone="Asia/Shanghai"),
    retries=2,
    retry_delay_seconds=300,
    log_prints=True
)
def daily_kpi(date: str = None):
    """计算每日 KPI 指标"""
    print(f"开始计算 {date or '今天'} 的 KPI")
    # 业务逻辑...
    pass

我的习惯：每次新建 flow 时，我会先花 5 分钟把配置写好。名字、描述、版本、标签、调度，一个不落。这 5 分钟的投资，会在后续的维护中节省你数小时的时间。

配置写好了，flow 就有了清晰的「身份」。不管是自己维护，还是交接给同事，都能快速上手。嗯，这就是优雅编排的第一步。