Apache Iceberg：历史数据加载时能否手动设置快照时间？

Iceberg完全支持保留原始变更时间和历史记录，核心是通过自定义快照时间戳、按原始时间顺序构建快照序列来实现，针对你的两种数据存储场景，具体操作如下：

1. 原表内嵌变更记录的迁移 #

如果原表本身存了每条数据的版本号、变更时间戳这类字段，按以下步骤来：

• 严格按照原始变更的时间先后顺序，分批导入每个版本的数据集：
  • 对每个历史版本的数据，用INSERT OVERWRITE（全量快照）或MERGE INTO（增量变更）导入Iceberg表，之后通过Iceberg的API把当前快照的时间戳改成原始变更时间。
  • Spark环境下的示例代码：

    import org.apache.iceberg.spark.SparkTable
    import org.apache.iceberg.util.DateTimeUtil
    
    val icebergTable = spark.table("target_iceberg_table").asInstanceOf[SparkTable].table()
    // 替换成当前批次数据对应的原始变更时间戳（毫秒）
    val originalChangeTime = 1620000000000L
    val snapshotTimestampMicros = DateTimeUtil.millisToMicros(originalChangeTime)
    
    // 导入数据后更新快照元数据
    val latestSnapshot = icebergTable.currentSnapshot()
    icebergTable.updateSnapshotMetadata(latestSnapshot.snapshotId(), meta => {
      meta.setTimestampMillis(originalChangeTime)
    })
    

• 全部导入后，就能通过Iceberg的时间旅行功能（比如SELECT * FROM target_iceberg_table TIMESTAMP AS OF '2021-05-03 12:00:00'）查询任意原始时间点的数据。

2. 审计表（如Hibernate Envers）的迁移 #

如果变更历史存在独立审计表，需要先把审计记录转换成Iceberg的快照序列：

• 先梳理审计表中的事件：区分新增、更新、删除操作，以及每个操作对应的原始时间戳。
• 按时间顺序依次重放这些操作到Iceberg表，每次操作完成后，用API把快照时间戳改成审计记录里的原始时间。
• 对于删除操作，直接用Iceberg的DELETE语句执行，之后同样调整快照时间，确保删除事件的时间和原始审计时间一致。

重要提醒 #

• 一定要按原始变更的时间顺序导入，不然Iceberg的快照时间线会乱，时间旅行查询会出错。
• 不同计算引擎（Spark、Flink、Trino）的API细节有差异，但核心都是修改快照的时间戳。
• 如果原始变更频率很高，建议合并相邻的小快照，避免Iceberg元数据太大影响性能。

内容的提问来源于stack exchange，提问作者Scott J.