如何在Prefect 2流运行时获取计划执行时间？

问题根源 #

你碰到的expected_start_time偏移问题，是因为这个字段会跟着重试、延迟执行的实际调度调整走，不是严格对应cron表达式定义的原始触发时间。

正确解决方法：用流运行的scheduled_start_time #

1. 代码实现 #

直接从流运行上下文里拿flow_run.scheduled_start_time，这个值就是cron定义的精确计划时间——哪怕重试、延迟执行，它都不会变。修改你的代码如下：

from prefect.context import get_run_context
from datetime import datetime, timedelta

@task(log_prints=True)
def calculate_time_bounds():
    context = get_run_context()
    # 拿到原始计划的调度时间
    scheduled_time = context.flow_run.scheduled_start_time
    print(f"原始计划执行时间: {scheduled_time}")
    
    # 按你的逻辑计算数据时间边界
    # 把计划时间的时分秒置零，得到当日00:00
    current_day_start = scheduled_time.replace(hour=0, minute=0, second=0, microsecond=0)
    # 减去1天，得到前一日的00:00
    previous_day_start = current_day_start - timedelta(days=1)
    
    print(f"数据查询起始: {previous_day_start}")
    print(f"数据查询结束: {current_day_start}")
    return previous_day_start, current_day_start

2. 效果验证 #

• 流准时在2024-05-15 02:00运行时，scheduled_start_time就是2024-05-15 02:00:00
• 要是流失败后03:00重试，这个值依然是2024-05-15 02:00:00，不会跟着实际执行时间变
• 手动回填过去某个cron时间的流运行时，scheduled_start_time会对应你指定的那个计划时间，不是当前时间

为啥task_run.expected_start_time不行？ #

这个是任务级别的预期启动时间，会受任务依赖、重试间隔、资源调度延迟等影响，所以会有秒级偏移，没法代表cron定义的原始流计划时间。而流级别的scheduled_start_time是严格绑定cron生成的触发时间，正好是你要的“原始计划时间”。

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