如何在Delta表执行Merge操作时手动指定版本关联时间戳？

Delta Lake默认用Merge操作的执行时间作为版本时间戳，但你完全可以通过以下两种方案实现将数据集关联到自定义业务时间戳，适配重入数据这类场景：

方案1：添加自定义业务时间戳字段（推荐） #

这是最直接且灵活的方式——在Delta表中新增一个业务时间戳字段（比如batch_timestamp），Merge时主动将该字段设置为你需要关联的时间（比如上周数据转储的时间）。后续查询时直接基于这个字段过滤，完全和操作时间解耦。

修改Merge代码示例： #

# 定义要关联的业务时间戳（比如上周数据转储时间）
target_batch_ts = "2022-11-07 00:00:00"

delta_table = DeltaTable.forPath(spark, delta_path)
delta_table.alias("t").merge(
  df.alias("s"), 
  "t.id = s.id"  # 关联条件
).whenMatchedUpdate(
  set={
    "col1": "s.col1",
    "col2": "s.col2",
    "batch_timestamp": target_batch_ts  # 更新时同步业务时间戳
  }
).whenNotMatchedInsert(
  values={
    "id": "s.id",
    "col1": "s.col1",
    "col2": "s.col2",
    "batch_timestamp": target_batch_ts  # 插入时设置业务时间戳
  }
).execute()

按业务时间查询示例： #

# 查询关联上周转储时间的所有数据
df = spark.read.format("delta")\
  .load("s3://BUCKET/PREFIX/")\
  .filter(f"batch_timestamp = '{target_batch_ts}'")

如果需要同时结合操作时间和业务时间，也可以叠加timestampAsOf和字段过滤，满足更复杂的查询需求。

方案2：手动映射版本号与业务时间 #

如果一定要复用Delta的版本时间戳机制，可以先执行Merge，再通过DESCRIBE HISTORY获取该版本的版本号，手动维护一个版本号-业务时间的映射表。后续查询时通过versionAsOf指定版本号，再关联映射表得到业务时间。

步骤示例： #

1. 执行Merge操作后，查询版本历史获取最新版本号：

history_df = delta_table.history(1)  # 获取最近1条版本记录
version_num = history_df.select("version").first()[0]

2. 将version_num和目标业务时间（如2022-11-07 00:00:00）存入映射表（比如另一个Delta表或数据库表）。
3. 查询时通过版本号关联：

# 通过版本号查询对应数据
df = spark.read.format("delta")\
  .option("versionAsOf", version_num)\
  .load("s3://BUCKET/PREFIX/")

这种方案适合必须依赖Delta版本机制的场景，但需要额外维护映射关系，灵活性不如方案1。

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