使用foreachBatch()的Spark流应用Driver OOM问题排查与优化

Spark无状态流应用中，Driver的元数据（如微批对应的Job、Stage、DataFrame lineage信息）默认会在微批执行完成，且相关对象无代码强引用时，由JVM GC自动回收。但如果存在未释放的全局变量、闭包引用，或Spark内部缓存的残留引用，会导致元数据无法被回收，进而累积占用内存。

可配置的激进清理策略：

• 调优JVM GC参数：通过spark.driver.extraJavaOptions启用G1GC并设置更早的回收触发阈值，例如：

  spark.driver.extraJavaOptions=-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=30 -XX:MaxGCPauseMillis=100
  

  让GC更早触发，及时回收闲置元数据。
• 启用Spark上下文清理器：确保spark.cleaner.referenceTracking默认开启，同时设置spark.cleaner.referenceTracking.cleanCheckpoints=true，让清理器主动清理过期的作业引用与临时文件。
• 限制微批元数据留存数量：设置spark.sql.streaming.minBatchesToRetain=10（默认100），减少Driver留存的最近微批元数据量，仅保留必要的历史批次信息。

即使是极小的DataFrame，频繁调用collect()仍可能导致Driver OOM：每次collect()会将全量数据拉取到Driver内存，虽然单批次数据量小，但每小时数百次的累积，加上若数据被全局变量持有（如存入List），会持续占用内存无法被GC回收。

替代方案：

• 下推计算到Executor：尽量避免在Driver处理数据，改用df.foreach(row => { /* 处理逻辑 */ })将计算放在Executor节点执行，减少Driver的数据拉取。
• 使用take(n)替代collect()：明确限制拉取行数，比如df.take(100)，避免意外拉取超出预期的数据量。
• 手动释放引用：拉取数据后立即处理，处理完成后将持有数据的变量置为null，帮助GC快速回收，示例：

  val data = df.collect()
  // 执行处理逻辑
  val result = processData(data)
  data = null // 手动释放引用
  

• 本地切断Lineage：对小DataFrame调用df.localCheckpoint()，切断其lineage，减少Driver存储的元数据量，同时数据临时存储在本地，避免长期占用内存。

不是必须完全避免缓存，但无状态应用中缓存使用不当会引发内存泄漏。无状态微批是独立的，跨微批的缓存毫无意义，反而会持续占用内存；若缓存后未及时释放，也会导致Driver/Executor内存累积。

优化缓存策略：

• 按需缓存+及时释放：仅在当前微批内确实需要多次复用的DataFrame才缓存，且在微批结束前必须调用df.unpersist(true)（true表示立即释放内存），示例：

  val cachedDF = df.cache()
  // 多次复用cachedDF的操作
  cachedDF.filter(...).write(...)
  cachedDF.join(...).write(...)
  cachedDF.unpersist(true) // 微批结束前释放
  

• 选择合适的存储级别：优先使用StorageLevel.MEMORY_ONLY（即cache()默认级别），小数据无需磁盘存储，避免额外IO与元数据开销。
• 避免跨微批缓存：不要在foreachBatch外部缓存DataFrame，无状态应用的微批独立，跨批缓存的数据无法复用，只会占用内存。
• 手动清除SQL缓存：若使用Spark SQL缓存表，可在每个微批结束时调用spark.sql("CLEAR CACHE")，清除当前会话的所有缓存表（注意仅适合微批完全独立的场景）。

除上述策略外，还可通过以下配置与实践优化Driver内存管理：

• 优化代码结构：避免在foreachBatch中使用全局变量存储数据或元数据，所有变量均设为局部变量，处理完成后自动释放引用；不要在外部集合（如List）中累积各微批的数据，避免内存持续增长。
• 减少元数据生成：合并多个DataFrame操作（如将连续的过滤、转换合并为一条操作链），减少不必要的中间DataFrame生成，从而降低lineage的复杂度与元数据量。
• 调整Driver内存开销：设置spark.driver.memoryOverhead=4g（默认是Driver内存的10%），给JVM非堆内存（用于存储元数据、线程栈等）预留足够空间，避免非堆内存溢出引发OOM。
• 禁用不必要的特性：关闭不需要的Spark功能，例如若无需内存 catalog，可设置spark.sql.catalogImplementation=hive；调整spark.sql.autoBroadcastJoinThreshold为合理值（如10MB），避免过大的广播表占用Driver内存。
• 监控与定位：通过Spark UI的Driver内存监控页，查看内存增长趋势，定位是元数据、缓存数据还是业务数据在累积；定期检查Storage页的缓存状态，确认缓存是否被及时释放。
• 利用Spark 3.x特性：启用自适应执行计划（spark.sql.adaptive.enabled=true），减少不必要的Stage生成，降低元数据量；设置spark.sql.streaming.execution.timeout=300s，避免卡住的微批长期占用元数据。

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