Databricks中Kafka流写入Delta表时Blob存储数据量异常激增

问题现象 #

在Databricks中通过Kafka流写入Delta表时，流入的数据未使表大小显著增长，但Blob存储中的数据量却大幅增加，两者差异达3-5倍。

核心原因分析 #

1. Delta Lake版本历史与旧文件堆积 #

Delta Lake默认保留7天的版本历史，每次merge（upsert）操作都会生成新的数据文件，旧文件仅被标记为逻辑删除而非物理删除。你的代码中使用whenMatched().updateAll()和whenNotMatched().insertAll()，每次微批都会对匹配记录全量更新，持续产生新文件，旧版本文件堆积导致存储占用远超当前表实际大小。

2. Checkpoint目录的状态数据积累 #

流处理配置的checkpointLocation目录会存储每个微批的状态数据、Kafka偏移量、元数据等信息。如果微批次数多或单次微批数据量大，该目录会积累大量状态文件，这部分数据不计入Delta表大小统计，却会占用Blob存储空间。

3. 微批分组逻辑的中间文件残留 #

代码中对system_device_id和uid_name做groupBy并取max(record)，该操作会产生shuffle中间文件。若存储层未及时清理临时文件，或Spark临时文件保留策略不当，也会导致额外存储占用（此为次要原因，通常为临时堆积）。

解决措施 #

清理Delta表旧版本文件 #

• 执行VACUUM命令物理删除旧版本文件，例如保留最近2天版本（根据业务需求调整）：

  VACUUM table_name RETAIN 2 DAYS;
  

• 修改Delta表配置，缩短版本保留周期：

  ALTER TABLE table_name SET TBLPROPERTIES (
    delta.logRetentionDuration = '2 days',
    delta.deletedFileRetentionDuration = '2 days'
  );
  

注意：执行VACUUM后无法恢复到更早版本，需确保业务无需历史版本回溯。

优化Checkpoint目录管理 #

• 定期清理冗余checkpoint数据：若作业为周期性运行（如使用Trigger.AvailableNow），可在作业完成后清理旧微批状态文件（需保留最新状态以支持后续续跑）；或为不同运行周期分配独立checkpoint目录，避免旧状态堆积。
• 调整微批大小，减少微批次数：通过maxFilesPerTrigger等参数控制单次微批处理数据量，降低状态文件生成频率。

优化Upsert逻辑 #

• 避免全量更新：当前代码中whenMatched().updateAll()会更新所有字段，可评估业务需求，仅更新实际变化的字段，减少每次更新产生的数据量。
• 增加merge过滤条件：在merge的ON条件中加入时间范围等过滤规则，减少每次需要匹配处理的记录数，降低写入压力。

清理Spark临时文件 #

• 手动清理默认临时文件目录：

  dbutils.fs.rm("dbfs:/tmp", recurse=true)
  

• 调整Spark临时文件保留策略，确保作业结束后自动清理临时文件。

内容的提问来源于stack exchange，提问作者Berkay Babataş