Spark Streaming有状态作业SST文件数量无限增长问题咨询

不正常。RocksDB原生自带Compaction机制，会自动合并SST文件、清理过期或冗余数据，但在S3这类对象存储环境下，由于对象存储的高延迟、原子性限制，再加上Spark配置不当，很容易导致Compaction逻辑失效或运行效率极低，最终出现SST文件持续堆积的情况。

• 调优RocksDB Compaction配置
  Spark提供了一系列针对RocksDB的配置参数，可直接优化Compaction行为：

  • 将spark.sql.streaming.stateStore.rocksdb.compaction.style设为LEVEL（默认可能是UNIVERSAL，LEVEL模式更适合长期运行的大规模状态存储，能更稳定地控制文件数量）
  • 根据你的状态数据量，调整spark.sql.streaming.stateStore.rocksdb.compaction.level.maxSize和spark.sql.streaming.stateStore.rocksdb.compaction.level.targetSize，设置合理的层级文件大小，减少小SST文件的生成
  • 开启spark.sql.streaming.stateStore.rocksdb.compaction.filter.deletes，自动清理已标记删除的状态条目（比如水位线超过1小时后，过期的去重键）

• 确保水位线过期状态清理生效
  你的作业基于1小时水位线去重，要保证Spark的状态过期机制正常工作：

  • 确认spark.sql.streaming.stateStore.minDeltasForSnapshot设置合理，避免因频繁生成快照导致大量小文件
  • 检查spark.sql.streaming.stateStore.cleanupDelay（默认1小时），确保过期状态数据被及时清理，减少Compaction需要处理的数据量

• 手动触发Compaction（应急处理）
  如果当前SST文件已经大量堆积，可以在作业运行期间，通过Spark API手动触发RocksDB Compaction（注意：需在Driver端执行，可能短暂影响作业性能）：

  import org.apache.spark.sql.streaming.StreamingQuery
  query.stores.foreach(_.asInstanceOf[org.apache.spark.sql.execution.streaming.state.RocksDBStateStore].compact())
  

• S3层小文件合并优化
  即使RocksDB层面优化到位，S3上的小文件仍可通过以下方式处理：

  • 利用Databricks的小文件合并工具，在作业暂停或低峰时段，将目录下的小SST文件合并为大文件后替换原目录（操作前务必备份状态数据）
  • 开启S3的Intelligent-Tiering存储类，自动将不常用的小文件归档到低成本存储层，降低存储成本压力

• 调整Checkpoint策略
  避免过于频繁生成Checkpoint，合理设置spark.sql.streaming.minBatchesToRetain，只保留必要的历史批次Checkpoint，减少冗余文件数量

RocksDB在对象存储上的Compaction效率天生不如本地磁盘，因为对象存储随机读写延迟高，所以配置调优是核心。另外，之前使用HDFS状态存储时的OOM问题，大概率是状态数据量超过内存阈值导致，可以尝试调小spark.sql.streaming.stateStore.memoryThreshold，让更多状态数据溢出到磁盘，或许能在不切换RocksDB的情况下解决OOM问题。

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