求助：Databricks中Spark Streaming RocksDB checkpoint上传无SST文件报错

报错日志 #

java.lang.IllegalStateException: Found no SST files during uploading RocksDB checkpoint version 498 with 2332 key(s).
    at com.databricks.sql.streaming.state.RocksDBFileManager.verifyImmutableFiles(RocksDBFileManager.scala:620)
    at com.databricks.sql.streaming.state.RocksDBFileManager.saveCheckpointToDbfs(RocksDBFileManager.scala:173)
    at com.databricks.sql.rocksdb.CloudRocksDB.$anonfun$sync$7(CloudRocksDB.scala:235)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
    at org.apache.spark.util.Utils$.timeTakenMs(Utils.scala:668)
    at com.databricks.sql.rocksdb.CloudRocksDB.timeTakenMs(CloudRocksDB.scala:634)
    at com.databricks.sql.rocksdb.CloudRocksDB.$anonfun$sync$1(CloudRocksDB.scala:234)
    at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
    at com.databricks.logging.UsageLogging.$anonfun$recordOperation$1(UsageLogging.scala:395)

背景 #

使用Spark 2.x默认配置在Databricks环境中做RocksDB有状态流处理，触发Checkpoint上传到S3时出现上述错误，临时解决方式是删除S3中的Checkpoint目录后重启流处理管道。

1. 错误成因：为什么找不到SST文件？ #

• RocksDB的SST文件是不可变的磁盘持久化文件，Checkpoint上传时需要扫描这些文件同步到远端存储。报错的核心是：当前Checkpoint版本（498）对应的本地RocksDB实例里，有2332条键值对，但没有生成可上传的SST文件。
• 具体场景包括：
  • 流处理的状态更新全部留在RocksDB的内存MemTable中，还没触发Flush到磁盘生成SST文件（默认触发条件是MemTable达到大小阈值或超时）。
  • 本地临时目录的SST文件被Databricks节点的存储回收机制误删。
  • RocksDB的Checkpoint逻辑异常，没有正确生成或识别SST文件。

2. Databricks中处理RocksDB Checkpoint的必要配置 #

默认配置存在缺陷，需要调整以下关键参数：

• spark.sql.streaming.stateStore.providerClass: 必须指定为com.databricks.sql.streaming.state.RocksDBStateStoreProvider（Spark 2.x中Databricks专属的RocksDB实现）。
• spark.sql.streaming.stateStore.rocksdb.memtableFlushThreshold: 降低MemTable刷新阈值，比如设为67108864（64MB），强制更早生成SST文件，避免状态全在内存。
• spark.sql.streaming.stateStore.rocksdb.localDir: 指定稳定的本地存储目录（比如节点的实例存储），别用默认临时目录，防止文件被自动清理。
• spark.sql.streaming.checkpointLocation: 确保S3路径有正确读写权限，且没有外部进程修改该目录下的文件。

3. 可行的解决方案与规避措施 #

• 调整RocksDB刷新策略：
  • 同时降低memtableFlushThreshold和memtableFlushIntervalMs参数，强制MemTable定期刷盘生成SST文件，避免状态长期驻留内存。
• 优化本地存储配置：
  • 把rocksdb.localDir配置到Databricks节点的持久化本地存储，而非临时目录，防止文件被回收。
• Checkpoint目录权限与隔离：
  • 确保流处理作业对S3的Checkpoint目录有独占读写权限，禁止其他作业或工具修改该目录下的文件。
• 作业重启策略优化：
  • 出现错误时，只删除无效的Checkpoint版本目录（而非整个Checkpoint目录），减少数据重算范围（可编写脚本识别错误版本并清理）。
• 升级Spark版本：
  • Spark 2.x的RocksDB集成bug较多，升级到Spark 3.x+配合Databricks Runtime 7.0+版本，官方修复了大量Checkpoint同步相关问题。

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