如何获取Delta Lake表两个版本的数据差异 开源实现方案咨询

以下是几种不依赖Databricks商业版功能的可行实现方式：

方案1：优化现有DataFrame对比逻辑，支持区分变更类型 #

你原有的实现只能拿到差异行集合，无法区分行是新增、删除还是修改，用全外连接可以实现更精准的差异识别：

import org.apache.spark.sql.functions._

// 此处替换为你实际的表主键字段，多主键可以传入Seq("id1", "id2")
val primaryKey = "id"

// 读取两个版本的数据并追加版本标识列
val dfV1 = spark.read.format("delta")
  .option("versionAsOf", 1)
  .load("/path/to/my/table")
  .withColumn("version", lit(1))

val dfV2 = spark.read.format("delta")
  .option("versionAsOf", 2)
  .load("/path/to/my/table")
  .withColumn("version", lit(2))

// 关联对比生成差异结果
val diffDF = dfV1.join(dfV2, Seq(primaryKey), "full_outer")
  .select(
    coalesce(dfV1(primaryKey), dfV2(primaryKey)).alias(primaryKey),
    when(dfV1(primaryKey).isNull, "新增")
      .when(dfV2(primaryKey).isNull, "删除")
      .when(lit(dfV1.drop("version") != dfV2.drop("version")), "修改")
      .alias("change_type"),
    struct(dfV1.drop(primaryKey).columns.map(c => dfV1(c).alias(c)): _*).alias("old_value"),
    struct(dfV2.drop(primaryKey).columns.map(c => dfV2(c).alias(c)): _*).alias("new_value")
  )
  .filter(col("change_type").isNotNull)

优势：不需要提前开启任何配置，适配所有Delta版本，适合小表或者偶发的差异查询场景。

方案2：使用开源Delta Lake自带的变更数据馈送功能 #

Delta Lake 2.0及以上版本已经将原本商业版的CDF（变更数据馈送）功能开源，不需要依赖Databricks商业服务即可使用。
使用前先开启表的CDF配置：

-- 新建表时开启
CREATE TABLE delta_table (id INT, value STRING)
USING DELTA
TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- 已有表开启
ALTER TABLE delta_table SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

开启后直接查询指定版本区间的变更数据即可：

val changeDF = spark.read.format("delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", 1)
  .option("endingVersion", 2)
  .load("/path/to/my/table")

返回结果会自带_change_type、_commit_version、_commit_timestamp三个元数据字段，直接标识每行的变更类型。
优势：性能远高于全量DataFrame对比，不需要自行写逻辑判断变更类型，适合频繁需要查询版本差异的生产场景。

方案3：解析Delta事务日志获取增量变更 #

如果你使用的Delta版本低于2.0，又不想做全量DataFrame对比，可以直接解析Delta表_delta_log目录下的事务日志文件，提取两个版本之间标记为add/remove的数据文件路径，直接读取对应文件即可拿到变更数据。该方案实现相对复杂，适合对Delta底层逻辑熟悉的开发者使用。

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