Spark SQL如何从Delta表实现批量读取与增量读取？

针对你需要从Delta表实现增量读取新增/更新数据的需求，这里提供几种实用的解决方案：

方法一：基于Delta版本号增量读取 #

Delta表会自动记录每个提交的版本号，通过追踪版本号可以精准获取增量数据：

1. 第一次批量读取时记录当前版本：
   执行SQL获取表的最新版本号，然后把这个值持久化存储（比如存到数据库、配置文件或缓存中）：

   # Python示例
   current_version = spark.sql("DESCRIBE HISTORY tablename").select("version").orderBy("version", ascending=False).first()[0]
   # 把current_version保存到外部存储，比如写入本地文件或Redis
   

2. 第二次增量读取：
   读取时指定从记录的版本号之后的内容：

   incremental_df = spark.read.format("delta").option("startingVersion", current_version + 1).table("tablename")
   

   这个方法会获取从current_version + 1版本开始的所有新增和更新数据。

方法二：基于时间戳增量读取 #

如果不需要精确到版本，也可以用时间戳来过滤增量：

1. 第一次读取时记录最新提交时间：
   获取Delta表最新的提交时间并存储：

   latest_commit_time = spark.sql("DESCRIBE HISTORY tablename").select("timestamp").orderBy("timestamp", ascending=False).first()[0]
   

2. 第二次增量读取：
   指定起始时间戳读取：

   incremental_df = spark.read.format("delta").option("startingTimestamp", latest_commit_time).table("tablename")
   

   另外，如果你的表本身带有业务时间戳字段（比如update_time），也可以直接用这个字段过滤：

   # 假设第一次读取时记录了max_update_time
   incremental_df = spark.table("tablename").filter(f"update_time > '{max_update_time}'")
   

方法三：使用Delta Change Data Feed（CDF） #

如果需要捕获所有变更（包括删除操作），可以开启Delta的CDF功能：

1. 先开启表的CDF属性：

   ALTER TABLE tablename SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
   

2. 记录起始版本/时间后，读取变更数据：

   # 基于版本读取变更
   incremental_df = spark.read.format("delta")\
       .option("readChangeFeed", "true")\
       .option("startingVersion", current_version)\
       .table("tablename")
   

   结果表会包含_change_type字段，标识该记录的变更类型：insert（新增）、update_postimage（更新后的数据）、update_preimage（更新前的数据）、delete（删除），你可以根据需求筛选需要的内容。

注意事项 #

• 版本号、时间戳这类追踪值必须持久化存储，不能仅存在内存中，否则程序重启后会丢失追踪点。
• 使用业务时间戳过滤时，要确保该字段会在数据更新时同步更新，否则会漏读数据。
• CDF适合需要完整变更链路的场景，普通版本/时间戳读取更适合只需要最新数据的场景。

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