如何在Dagster中创建带调度的日期分区以支持数据回溯？

你的get_date_partitions函数实现逻辑错误：当前你将所有日期的列表传入单个Partition实例的构造参数，再把这一个Partition放入返回的列表中，Dagster会识别为你只有1个包含全部日期的分区，自然会出现所有日期合并展示的问题。

1. 修复分区生成函数 #

你需要为每个日期单独生成Partition实例，同时注意不要在模块顶层定义TODAY变量，否则服务启动后TODAY会一直保留启动时的日期不会自动更新，修正后代码如下：

DATE_FORMAT = "%Y-%m-%d"
BACKFILL_DATE = "2021-04-01"

def get_number_of_days():
    # 将TODAY移入函数内部，每次调用都会获取最新日期
    TODAY = datetime.today()
    backfill_date_obj = datetime.strptime(BACKFILL_DATE, DATE_FORMAT)
    delta = TODAY - backfill_date_obj
    return delta.days

def get_date_partitions():
    TODAY = datetime.today()
    # 每个日期对应一个Partition实例，放在同一个列表中返回
    return [
        Partition(datetime.strftime(TODAY - timedelta(days=x), DATE_FORMAT))
        for x in range(get_number_of_days())
    ]

2. 修复分区选择器逻辑 #

你当前的分区选择器直接返回全部分区，会导致每天调度时触发所有历史分区的执行，需要调整为仅返回调度时间对应的日期分区：

def daily_partition_selector(
    ctx: ScheduleExecutionContext, partition_set: PartitionSetDefinition
) -> Union[Partition, List[Partition]]:
    scheduled_date = ctx.scheduled_execution_time.strftime(DATE_FORMAT)
    # 找到对应日期的分区返回
    for partition in partition_set.get_partitions():
        if partition.value == scheduled_date:
            return partition
    # 兼容异常情况，返回空列表不执行
    return []

更简便的官方原生方案 #

Dagster已经内置了按天分区的实现DailyPartitionsDefinition，不需要手写分区生成逻辑，兼容自动回溯、调度逻辑也不需要自定义选择器，实现代码更简洁不易出错：

from dagster import DailyPartitionsDefinition, build_schedule_from_partitioned_job

# 直接定义按天分区，指定开始回溯的日期即可
date_partitions_def = DailyPartitionsDefinition(start_date="2021-04-01")

# 绑定到你的Job/Pipeline上后，直接用内置方法生成调度
my_schedule = build_schedule_from_partitioned_job(
    job=my_pipeline,
    name="my_schedule",
    cron_schedule="15 8 * * *",
    execution_timezone="UTC",
)

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