Spark每日Delta Lake数据处理任务：如何避免重复处理历史数据？

针对你遇到的每日任务重复处理历史数据的问题，这里有几个实用的解决思路：

1. 利用Delta Lake的增量版本读取特性 #

Delta Lake会记录每一次数据变更的版本号，你可以跟踪每次任务处理到的版本，下次直接读取该版本之后的新增/更新数据，完全跳过已处理过的历史数据。

• 操作步骤：
  1. 把上次任务处理完成时的Delta表版本号存在外部存储（比如数据库、本地配置文件）
  2. 本次任务读取时，指定从该版本开始的增量数据
  3. 处理完成后，更新存储的版本号为当前表的最新版本
• 示例代码：

// 从外部存储获取上次处理的版本号
val lastProcessedVersion = 123L
// 读取增量数据
val deltaDF = spark.read.format("delta")
  .option("startingVersion", lastProcessedVersion)
  .load("/path/to/delta-table")
// 获取当前表的最新版本
val currentVersion = spark.sql("DESCRIBE HISTORY delta.`/path/to/delta-table`")
  .select("version").head().getLong(0)
// 将currentVersion写入外部存储，供下次任务使用

2. 在Delta表中添加已处理标记字段 #

给Delta表新增一个is_processed布尔字段，初始值设为false。每日任务只筛选is_processed = false且时间戳在最近7天的数据，处理完成后批量把这些数据的标记改为true。

• 注意：更新操作要保证原子性，避免多任务并发时重复处理，可结合Delta的事务特性来实现。
• 示例代码：

// 读取未处理的最近7天数据
val unprocessedDF = spark.read.format("delta")
  .load("/path/to/delta-table")
  .where("is_processed = false AND event_timestamp >= date_sub(current_date(), 7)")

// 执行数据处理逻辑...

// 标记已处理数据
import io.delta.tables._
val deltaTable = DeltaTable.forPath(spark, "/path/to/delta-table")
deltaTable.update(
  condition = expr("event_timestamp >= date_sub(current_date(), 7) AND is_processed = false"),
  set = Map("is_processed" -> lit(true))
)

3. 转为流处理+水印机制 #

如果业务场景允许，把批处理改成Structured Streaming任务。流任务会自动维护检查点记录已处理的位置，再配合水印设置7天的延迟窗口，既能处理延迟数据，又不会重复处理历史数据。

• 示例代码：

val streamDF = spark.readStream.format("delta")
  .load("/path/to/delta-table")
  // 设置7天水印，自动清理过期的状态数据
  .withWatermark("event_timestamp", "7 days")
  // 这里替换成你的实际处理逻辑
  .groupBy(window($"event_timestamp", "1 day"), $"user_id")
  .agg(count("*").alias("cnt"))

// 写入目标数据库
streamDF.writeStream
  .format("jdbc")
  .option("url", "jdbc:mysql://host:port/db_name")
  .option("dbtable", "target_table")
  // 检查点路径，用于记录已处理位置
  .option("checkpointLocation", "/path/to/checkpoint_dir")
  .start()
  .awaitTermination()

4. 记录上次任务的处理时间戳 #

在外部存储中保存每次任务完成的时间戳，本次任务只读取时间戳大于上次处理时间且在最近7天范围内的数据。这样既覆盖了延迟数据，又跳过了之前已经处理过的历史数据。

• 注意：如果任务失败，要回滚时间戳，避免遗漏数据。
• 核心筛选条件示例：

event_timestamp >= date_sub(current_date(), 7) 
AND event_timestamp > '上次任务处理完成时间'

内容的提问来源于stack exchange，提问作者khÜs h