如何在Dagster资产/作业中指定并使用幂等执行日期？

Dagster没有Airflow那样的Jinja模板语法，但可以通过分区资产和运行上下文实现完全等效的日期绑定与幂等性保障，核心是把执行日期作为资产的分区键，确保每次运行都针对固定日期的数据集。

1. 定义带日期分区的核心资产 #

用DailyPartitionsDefinition创建按天分区的规则，每个资产实例对应一个具体日期。通过上下文获取分区键后，可转换为无横杠格式（对应Airflow的{{ds_nodash}}）：

from dagster import DailyPartitionsDefinition, asset
import os

# 定义按天分区的规则，起始日期按需设置
daily_partitions = DailyPartitionsDefinition(start_date="2022-01-01")

@asset(partitions_def=daily_partitions)
def processed_daily_file(context):
    # 获取当前运行的分区日期（格式：YYYY-MM-DD）
    partition_date = context.partition_key
    # 转换为无横杠格式，匹配你的文件名规则
    ds_nodash = partition_date.replace("-", "")
    target_file = f"FILE_{ds_nodash}.csv"

    # 读取并处理文件逻辑
    if not os.path.exists(target_file):
        raise FileNotFoundError(f"目标文件 {target_file} 不存在")
    
    # 示例处理逻辑：读取CSV、清洗数据
    import pandas as pd
    df = pd.read_csv(target_file)
    cleaned_df = df.dropna()

    # 将结果写入与分区绑定的位置（保证幂等性：重复运行会覆盖或复用该结果）
    output_path = f"processed_FILE_{ds_nodash}.parquet"
    cleaned_df.to_parquet(output_path)
    return output_path

2. 传感器中关联分区触发运行 #

如果需要像Airflow传感器那样检测文件存在性再触发运行，Dagster的传感器可以结合分区机制，只触发未处理日期的资产运行：

from dagster import SensorResult, RunRequest, sensor

@sensor(asset_selection=[processed_daily_file])
def daily_file_sensor(context):
    # 获取所有未物化（未处理）的分区日期
    unprocessed_partitions = context.instance.get_unmaterialized_partitions(
        processed_daily_file.key,
        daily_partitions
    )
    run_requests = []
    for partition_key in unprocessed_partitions:
        ds_nodash = partition_key.replace("-", "")
        target_file = f"FILE_{ds_nodash}.csv"
        # 检查文件是否存在
        if os.path.exists(target_file):
            run_requests.append(
                RunRequest(
                    partition_key=partition_key,
                    run_key=f"daily_file_{ds_nodash}"
                )
            )
    return SensorResult(run_requests=run_requests)

3. 保障幂等性的关键细节 #

• 绑定分区到输出：处理后的结果必须与分区日期关联（比如带日期后缀的文件、数据库分区），重复运行同一分区时，要么覆盖原有结果，要么根据物化状态跳过运行。
• 固定日期来源：所有依赖日期的逻辑必须通过context.partition_key获取，禁止使用系统当前时间，避免重复运行时日期偏移。
• 利用物化状态跟踪：Dagster会自动记录每个分区资产的物化状态，已物化的分区默认不会重复执行（除非手动触发重新物化），从机制上保证同一日期的数据无论运行多少次，处理逻辑和结果都一致。

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