基于Spark+Kafka+Iceberg的实时流系统：如何减少快照生成数量

以下是几个可行的方案，均无需调整批处理记录数，同时能保留实时处理能力：

1. 启用Iceberg自动合并小文件/快照功能 #

Iceberg支持在提交时自动合并符合条件的小文件和快照，无需额外批处理任务，也不会改变流处理的批大小。

配置方式 #

在SparkSession初始化时添加以下配置：

spark.conf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
spark.conf.set("spark.sql.iceberg.commit.merge.enabled", "true")
spark.conf.set("spark.sql.iceberg.commit.merge.min-files", "5")  // 当分区内小文件数超过5个时触发合并
spark.conf.set("spark.sql.iceberg.commit.merge.max-files", "20") // 单次合并最多处理20个文件
spark.conf.set("spark.sql.iceberg.commit.merge.target-file-size-bytes", "134217728") // 合并后目标文件大小为128MB

原理 #

每次流处理提交时，Iceberg会自动检查当前分区的文件数量，若达到阈值则合并小文件，并生成一个新的快照替代多个小快照，从而减少总快照数量。整个合并过程在提交阶段完成，不影响实时数据摄入的批处理逻辑。

2. 配置快照自动过期策略 #

通过设置Iceberg表的快照过期规则，自动清理旧的、不再需要的快照，从而控制快照总数。

配置方式 #

通过SQL语句修改表属性：

ALTER TABLE your_catalog.your_db.your_table SET TBLPROPERTIES (
  'iceberg.table.snapshots.expire.min-age' = '86400000',  -- 快照至少保留1天（毫秒）
  'iceberg.table.snapshots.expire.max-snapshots' = '10', -- 最多保留10个最新快照
  'iceberg.table.snapshots.expire.interval-ms' = '3600000' -- 每小时自动检查并清理过期快照
)

原理 #

Iceberg会定期（按配置的间隔）自动清理超过保留时长或超出数量上限的旧快照。这个操作是后台异步执行的，完全不影响实时流的批处理记录数和处理延迟。

3. 优化Spark流提交与Iceberg写入模式 #

调整流处理的提交触发逻辑，结合Iceberg的事务性提交特性，减少不必要的快照生成。

配置方式 #

保持批处理记录数不变的前提下，设置以下参数：

// 启用Iceberg流处理的合并提交模式
spark.conf.set("spark.sql.iceberg.streaming.commit.mode", "append")
spark.conf.set("spark.sql.iceberg.streaming.commit.allow-overwrite", "false")
// 设置触发间隔（根据业务实时性需求调整，比如1分钟）
val streamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-servers")
  .option("subscribe", "your-topic")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .writeStream
  .format("iceberg")
  .option("path", "your-iceberg-table-path")
  .trigger(Trigger.ProcessingTime("1 minute"))
  .start()

原理 #

通过固定处理时间触发提交，避免过于频繁的小批量提交生成过多快照。同时Iceberg的append模式会将同批次的数据写入到对应分区的文件中，减少碎片化的快照生成，且不改变批处理的记录数。

4. 合理设置Iceberg表的分区与分桶 #

通过分区和分桶优化数据组织方式，让同一批次的写入尽可能合并到更少的文件和快照中。

配置方式 #

创建Iceberg表时指定分区和分桶：

CREATE TABLE your_catalog.your_db.your_table (
  id STRING,
  content STRING,
  event_time TIMESTAMP
)
PARTITIONED BY (DATE(event_time))
CLUSTERED BY (id) INTO 10 BUCKETS
STORED BY ICEBERG

原理 #

• 分区：按时间（如小时/天）分区后，同一时间窗口的流数据会写入同一个分区，减少跨分区的快照数量。
• 分桶：相同分桶键的数据会被写入到同一个文件，避免生成大量小文件，从而减少对应的快照数量。
  这些优化都是在写入阶段自动完成的，不需要调整批处理记录数，也不会影响实时处理能力。

内容的提问来源于stack exchange，提问作者Sơn Bùi