如何在Prefect 1.x中设置每月首个周二18:00 IST及自定义日期的Flow调度？

一、每月第一个周二18:00 IST 调度实现 #

由于Cron无法直接表达「每月第一个周二」这类规则，推荐使用RRuleSchedule（基于iCalendar的RRule标准）来实现，这是Prefect 1.x中处理复杂重复调度最可靠的方式。

代码示例 #

from prefect import Flow, task
from prefect.schedules import RRuleSchedule
import rrule as rr
import pytz
import datetime

# 定义IST时区
ist_tz = pytz.timezone("Asia/Kolkata")

# 构建RRule规则：每月第一个周二的18:00 IST
rrule_schedule = RRuleSchedule(
    rrule=rr.rrule(
        rr.MONTHLY,
        byweekday=rr.TU(1),  # TU(1)表示每月第一个周二
        byhour=18,
        byminute=0,
        bysecond=0,
        dtstart=ist_tz.localize(datetime.datetime(2024, 1, 1, 18, 0, 0)),  # 起始日期需匹配规则
        tzinfo=ist_tz
    )
)

@task
def sample_task():
    print("执行任务：每月第一个周二18:00 IST")

with Flow("每月第一个周二调度", schedule=rrule_schedule) as flow:
    sample_task()

# 注册或运行Flow
flow.register(project_name="your-project")

备选方案：每日调度加运行判断 #

如果不想依赖RRule，也可以用IntervalSchedule设置每日运行，然后在任务开头添加判断逻辑，仅当当天符合规则时才执行核心逻辑：

from prefect import Flow, task
from prefect.schedules import IntervalSchedule
from datetime import timedelta
import pytz
import datetime

ist_tz = pytz.timezone("Asia/Kolkata")
daily_schedule = IntervalSchedule(interval=timedelta(days=1), start_date=ist_tz.localize(datetime.datetime(2024, 1, 1, 18, 0, 0)))

@task
def check_and_run():
    today = datetime.datetime.now(tz=ist_tz)
    # 判断是否为每月第一个周二：周二对应weekday=1，每月前7天的第一个周二必在其中
    if today.weekday() == 1 and today.day <=7:
        print("执行任务：每月第一个周二18:00 IST")
    else:
        print("今日不符合调度规则，跳过任务")

with Flow("每日检查调度", schedule=daily_schedule) as flow:
    check_and_run()

flow.register(project_name="your-project")

二、自定义运行日期组的实现 #

如果需要指定一组固定的运行日期（例如2024-05-10、2024-06-15等），可以直接使用Schedule结合DateTimeClock来实现，每个自定义日期对应一个时钟。

代码示例 #

from prefect import Flow, task
from prefect.schedules import Schedule, DateTimeClock
import pytz
import datetime

ist_tz = pytz.timezone("Asia/Kolkata")

# 定义自定义运行日期（需带时区）
custom_dates = [
    ist_tz.localize(datetime.datetime(2024, 5, 10, 18, 0, 0)),
    ist_tz.localize(datetime.datetime(2024, 6, 15, 9, 30, 0)),
    ist_tz.localize(datetime.datetime(2024, 7, 20, 14, 0, 0))
]

# 将每个日期转换为DateTimeClock
clocks = [DateTimeClock(dt=date) for date in custom_dates]

# 创建调度
custom_schedule = Schedule(clocks=clocks)

@task
def custom_date_task():
    print("执行自定义日期任务")

with Flow("自定义日期调度", schedule=custom_schedule) as flow:
    custom_date_task()

flow.register(project_name="your-project")

注意事项 #

• 所有日期必须指定正确的时区（如IST），避免时间偏移问题；
• 如果自定义日期较多，可以从文件或数据库读取日期列表，动态生成DateTimeClock。

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