使用Spark SQL读取Iceberg表增量数据的问题咨询

问题1：为何未得到期望的3条记录（含更新的1条） #

Iceberg通过start-snapshot-id和end-snapshot-id进行的增量读取，默认仅捕获append操作新增的行数据。而你提到的overwrite操作本质是文件级的替换，并非行级更新——Iceberg不会将这类操作中的变更行识别为增量数据，因为快照间的差异计算是基于文件粒度，而非行粒度，所以只会返回append的2条记录，不会包含overwrite里的更新行。

问题2：如何读取更新的记录（用于维度表） #

要捕获包含更新、删除的行级变更，需要启用Iceberg的CDC（变更数据捕获）功能，具体步骤如下：

1. 确保目标Iceberg表开启CDC：建表时添加属性TBLPROPERTIES ('iceberg.cdc.enabled'='true')，已有表可通过ALTER TABLE修改属性。
2. 使用CDC模式读取增量变更，代码示例：

cdc_df = spark.read.format("iceberg") \
    .option("start-snapshot-id", str(first_snapshot_id)) \
    .option("end-snapshot-id", str(last_snapshot_id)) \
    .option("read-change-data", "true") \
    .load(table)

3. CDC结果包含_change_type字段，取值有insert、delete、update_before、update_after，筛选insert和update_after即可获取维度表需要的新增及更新后的记录：

dim_sync_df = cdc_df.filter("_change_type IN ('insert', 'update_after')")

问题3：当前读取Iceberg表的方式是否正确（后续将写入普通Hive分区格式） #

当前的基础增量读取方式本身是正确的，但仅适用于捕获append类型的新增数据。如果你的场景需要覆盖更新、删除类的变更，就需要切换到上述的CDC读取方式。

写入普通Hive分区格式时，注意两点：

• 读取到目标数据后，按Hive表的分区字段做分区写入：

dim_sync_df.write.mode("append") \
    .partitionBy("your_hive_partition_col") \
    .format("parquet") \
    .saveAsTable("your_hive_db.your_hive_table")

• 维度表同步建议用merge操作避免重复数据，可通过Spark的mergeInto语法实现，或先清理对应旧数据再插入新数据。

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