如何在Prefect V2中运行本地定时流？是否必须使用Deployment？

核心问题解答 #

1. 是否必须使用Deployment实现本地定时流？
   是的。Prefect V2中，持久化的定时调度逻辑与Deployment绑定——直接调用flow()或flow.run()仅会触发单次运行，无法自动重复执行。只有通过Deployment才能将流与调度规则、执行配置绑定，由Prefect默认的本地SQLite后端管理调度周期。

2. 是否存在无需UI/服务器的本地定时Deployment？
   存在。可以通过本地SQLite后端+本地Agent实现完全本地化的定时调度，无需启动Prefect Server或UI界面，所有调度逻辑都在本地运行。

具体实现步骤 #

第一步：修正基础流代码

先确保流代码能正常运行，给date参数添加默认值避免调用报错：

from prefect import flow, task
from datetime import datetime

@task
def hello_world():
    print("hello world")

@flow(flow_run_name="{name}-on-{date:%A}")
def flow_main(name="flow_main", date: datetime = None):
    # 若未传入date参数，默认使用当前UTC时间
    if date is None:
        date = datetime.utcnow()
    hello_world()

# 单次运行测试（可选）
if __name__ == "__main__":
    flow_main()

第二步：创建本地Deployment

你可以选择命令行方式或代码方式创建Deployment：

方式1：命令行创建（推荐）

假设你的代码文件名为flow_script.py，执行以下命令生成Deployment配置文件：

# 构建Deployment，指定cron调度、工作队列
prefect deployment build ./flow_script.py:flow_main --name "local-5min-schedule" --cron "*/5 * * * *" --work-queue "local-worker-queue"

执行后会生成flow_main-deployment.yaml文件，包含调度和执行配置。

接着应用这个Deployment：

prefect deployment apply flow_main-deployment.yaml

方式2：代码方式创建

如果偏好代码控制，可编写Deployment创建脚本：

from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
from flow_script import flow_main

# 构建并应用Deployment
deployment = Deployment.build_from_flow(
    flow=flow_main,
    name="local-5min-schedule",
    schedule=CronSchedule(cron="*/5 * * * *"),
    work_queue_name="local-worker-queue"
)

if __name__ == "__main__":
    deployment.apply()

运行该脚本即可完成Deployment的创建。

第三步：启动本地Agent

最后启动本地Agent，监听指定的工作队列，负责触发并执行定时流：

prefect agent start --work-queue "local-worker-queue"

运行说明 #

• Agent启动后，会每隔5分钟自动触发一次flow_main的运行，所有日志和执行记录都保存在本地~/.prefect目录下的SQLite数据库中。
• 全程无需启动Prefect Server或UI界面，完全本地化运行。

内容的提问来源于stack exchange，提问作者Matthieu Allier