如何处理源Delta表被重新创建后的批流同步问题？

当Delta源表被覆盖后，流作业checkpoint中记录的历史版本会因源表版本链被截断而失效，导致Trigger.AvailableNow()同步失败。以下是两种适配Spark 3.3.2版本的可靠解决思路：

方法一：版本检测+重置同步状态 #

通过对比源表的最早可用版本变化判断是否发生覆盖，自动截断目标表并重置流作业状态：

步骤1：存储源表基准版本 #

创建元数据表记录源表初始最早版本，用于后续对比校验：

// 初始化元数据表（仅首次运行执行）
spark.sql("CREATE TABLE IF NOT EXISTS table_metadata (table_name STRING, last_earliest_version BIGINT) USING delta")

步骤2：检测源表覆盖状态 #

每次启动同步前，检查源表当前最早版本与历史记录的差异：

// 获取源表Delta日志实例
val sourceDeltaLog = DeltaLog.forTable(spark, "path/to/source_table")
// 查询源表当前最早可用版本
val currentEarliestVersion = sourceDeltaLog.history.select(min("version")).head().getLong(0)

// 读取历史基准版本
val lastVersionOpt = spark.sql("SELECT last_earliest_version FROM table_metadata WHERE table_name = 'source_table'").headOption

lastVersionOpt match {
  case Some(row) =>
    val lastVersion = row.getLong(0)
    // 当前最早版本大于历史记录，说明源表被覆盖
    if (currentEarliestVersion > lastVersion) {
      // 截断目标表
      spark.sql("TRUNCATE TABLE target_table")
      // 删除流作业checkpoint目录（需确保HDFS权限）
      import org.apache.hadoop.fs.{FileSystem, Path}
      val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
      fs.delete(new Path("path/to/checkpoint"), true)
      // 更新基准版本记录
      spark.sql(s"UPDATE table_metadata SET last_earliest_version = $currentEarliestVersion WHERE table_name = 'source_table'")
    }
  case None =>
    // 首次运行，插入初始版本记录
    spark.sql(s"INSERT INTO table_metadata VALUES ('source_table', $currentEarliestVersion)")
}

步骤3：启动同步作业 #

使用Trigger.AvailableNow()启动批流同步：

val streamQuery = spark.readStream
  .format("delta")
  .load("path/to/source_table")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "path/to/checkpoint")
  .trigger(Trigger.AvailableNow())
  .start("path/to/target_table")

streamQuery.awaitTermination()

方法二：使用ignoreChanges选项跳过版本校验 #

Spark 3.2+支持ignoreChanges读取选项，当源表发生覆盖、更新时，流作业会跳过版本校验并重新读取全量数据，结合overwrite模式实现自动全量同步：

val streamQuery = spark.readStream
  .format("delta")
  .option("ignoreChanges", "true") // 忽略源表版本变化，重新读取全量数据
  .load("path/to/source_table")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "path/to/checkpoint")
  .option("mergeSchema", "true") // 可选：兼容源表schema变更
  .mode("overwrite") // 覆盖目标表，避免重复数据
  .trigger(Trigger.AvailableNow())
  .start("path/to/target_table")

streamQuery.awaitTermination()

注意事项 #

• ignoreChanges会增加作业开销，适合源表覆盖操作不频繁的场景；
• 方法一中的截断表与删除checkpoint操作需保证顺序执行，避免数据不一致；
• 若需兼顾增量同步与覆盖后的全量同步，可结合两种方法：平时用增量append模式，检测到覆盖时切换为overwrite模式。

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