Spark Streaming读写Delta表性能异常及缓存RDD问题咨询

环境配置：Spark 3.2.1、Delta 1.2.1、Scala 2.12.12、Hadoop 3.3.0
业务场景：从Gen1 Delta表读取流数据，处理后写入Gen2 Delta表
现象：多数流处理批次耗时不足1分钟，但每30分钟左右会出现一批次耗时约14分钟的情况。在SparkUI的Storage标签页中发现两个缓存的RDD：

Delta Table State #116243 - wasbs://INPUT_PATH/_delta_log  Disk Memory Serialized 1x Replicated 3.4 GiB (Size in Memory)
Delta Table State #60042 - abfss://OUTPUT_PATH/_delta_log Disk Memory Serialized 1x Replicated 55.7 GiB (Size in Memory)

疑问：这些缓存RDD及其大小是否存在问题？

1. 输出表的Delta状态缓存大小严重异常
   55.7 GiB的Delta表状态缓存是明显有问题的。Delta Lake的_delta_log目录存储的是事务元数据（JSON日志文件+周期性生成的Parquet checkpoint文件），正常情况下元数据体积远小于业务数据，即使是长期运行的流任务，元数据也不该达到几十GiB的规模。

2. 异常的可能原因

   • Delta日志未自动清理：Delta 1.2.1默认不会自动清理旧的日志文件，如果流任务长期运行且频繁写入，日志文件会持续积累，导致元数据体积暴增。每次流批次处理时，Spark需要加载全量的Delta表状态，大体积的元数据会导致加载、序列化/反序列化耗时剧增，这就是你看到每30分钟左右出现长耗时批次的原因（可能是触发了全量状态加载）。
   • Checkpoint状态积累：流任务的checkpoint中可能积累了过多的Delta表状态快照，导致每次恢复或状态刷新时需要处理大量数据。
   • 版本兼容性问题：你使用的Delta 1.2.1是较老的版本，旧版本在元数据管理、状态缓存优化上存在不足，比如缺乏高效的状态增量更新机制。

3. 解决建议

   • 启用Delta日志自动清理：在写入Delta表时配置delta.logRetentionDuration（比如设置为interval 7 days）和delta.deletedFileRetentionDuration，让Delta自动清理过期的日志和删除文件。可以通过SQL语句修改表属性：

     ALTER TABLE your_output_table SET TBLPROPERTIES (
       'delta.logRetentionDuration' = 'interval 7 days',
       'delta.deletedFileRetentionDuration' = 'interval 1 days'
     )
     

   • 手动清理旧日志：如果日志已经积累过多，可以使用Delta的VACUUM命令手动清理：

     VACUUM your_output_table RETAIN 7 DAYS
     

     注意：执行VACUUM前确保没有旧的流任务或批任务需要读取已删除的日志文件。
   • 升级Delta版本：升级到Delta 2.x系列版本，新版本对元数据管理、流处理状态优化有很大提升，能有效减少状态缓存的体积和加载耗时。
   • 检查流任务配置：确认流任务的checkpointLocation路径没有异常积累，必要时可以清理旧的checkpoint（注意：清理checkpoint会导致流任务从头开始处理，需谨慎操作）。
   • 验证实际日志目录大小：直接查看Gen2存储中abfss://OUTPUT_PATH/_delta_log的实际占用空间，确认缓存大小是否与实际目录大小匹配，排除SparkUI显示异常的可能。

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