Spark Structured Streaming在Kubernetes环境中执行器内存溢出问题排查求助

看起来你遇到了Spark Structured Streaming在Kubernetes环境下的执行器内存溢出（OOM）问题，结合你提供的作业细节、错误日志和堆dump信息，我来帮你拆解排查方向和可能的解决方案：

一、先理清OOM与FetchFailedException的关联 #

你看到的FetchFailedException其实是执行器OOM后的连锁反应：当某个执行器因为内存不足崩溃后，依赖它的Shuffle任务就会因为无法获取数据而失败，最终导致整个Stage重试多次后失败。所以核心问题还是要解决执行器的内存溢出。

二、State大小的计算与优化（堆dump显示State是最大对象） #

你的作业使用了滑动窗口、Watermark和多字段聚合，State里存储的是每个分组在未过期窗口内的聚合中间状态，这是内存占用的核心来源。

1. 如何估算State大小？ #

你可以通过以下思路大致估算：

• 先确定分组Key的基数：比如groupby(x)中的x是用户ID、设备ID这类高基数字段，还是低基数的类别字段？如果是高基数（百万级以上），每个Key对应一组窗口状态，总内存会快速膨胀。
• 再计算每个Key对应的窗口数量：你的回溯窗口是24小时，滑动间隔1小时，Watermark是1小时。理论上，Watermark会自动清理事件时间早于当前处理时间 - 1小时的窗口状态，所以每个Key实际保留的窗口数大约是23个（24小时回溯 - 1小时Watermark延迟）。
• 最后估算单个窗口的聚合数据大小：你有22个聚合特征，每个特征的大小（比如整数4字节、字符串平均长度等）乘以每个窗口内的记录数，就是单个窗口的内存占用。

2. 优化State内存占用的关键操作 #

• 验证Watermark是否生效：检查你的事件时间字段是否正确解析（确认timestampFormat与数据中的时间格式完全匹配，有没有出现解析错误导致所有事件时间一致的情况）。如果Watermark没生效，State会无限累积数据，必然OOM。可以在Spark UI的Streaming标签下查看State Size的变化趋势，如果持续增长不下降，说明Watermark没起作用。
• 调整Watermark延迟时间：如果你的数据实际延迟超过1小时，Watermark会无法及时清理过期状态，导致内存堆积。可以根据数据的实际延迟情况，适当调大Watermark（比如设为2小时），但不要超过回溯窗口的范围。
• 优化分组Key的基数：如果Key的基数过高，考虑是否可以通过预聚合、维度下钻等方式减少分组数量，或者采用加盐（Salting）的方式分散高基数Key的压力（避免单个执行器承载过多Key的State）。
• 调整Spark内存分配比例：默认情况下，Spark执行器堆内内存的60%用于存储（包括State），剩下40%用于任务执行。你可以通过spark.executor.memoryFraction参数提高存储内存的比例（比如设为0.7），给State更多内存空间，但要注意预留足够内存给任务执行，避免新的OOM。

三、AWS SDK与Hadoop-AWS的线程泄漏排查 #

你使用的aws-sdk-java-1.11.0版本确实存在一些已知的线程池泄漏问题，尤其是在频繁创建客户端实例的场景下：

• 检查客户端复用配置：确保Hadoop S3客户端启用了连接池复用，设置以下参数：

  spark.hadoop.fs.s3a.connection.maximum=30  # 调整为合适的连接数，默认15
  spark.hadoop.fs.s3a.connection.threadpool.size=64  # 控制线程池大小
  spark.hadoop.fs.s3a.connection.establish.timeout=5000
  spark.hadoop.fs.s3a.connection.timeout=10000
  

• 升级依赖版本：你的aws-sdk-java-1.11.0版本过于陈旧，很多线程泄漏的bug已经在后续版本修复。建议升级到与hadoop-aws-3.3.0匹配的AWS SDK版本（hadoop-aws-3.3.0官方依赖的是aws-sdk-java-1.11.1021），避免版本不兼容导致的问题。
• 排查自定义客户端逻辑：如果你的作业中有自定义的S3客户端创建逻辑，确保每次使用后正确关闭客户端，避免线程池资源泄漏。

四、其他排查与优化点 #

• 检查数据倾斜：查看Spark UI的Stage标签，观察每个任务的输入数据量，如果某个任务的输入远大于其他任务，说明存在数据倾斜。可以通过加盐分组、预聚合等方式解决，避免单个执行器承载过多数据。
• 优化ForeachBatch逻辑：
  • 避免在foreachBatch中使用全局变量（虽然你的场景下问题不大，但可能引发并发风险）；
  • 每次写入Parquet时，合理设置分区数（比如通过repartition或coalesce调整），避免小文件过多或单个文件过大；
  • 在日志中添加agg_df.count()、agg_df.rdd.getNumPartitions()等信息，方便监控每个批次的数据量和分区情况。
• 调整执行器资源配置：如果上述优化后内存仍然不足，可以考虑增加执行器数量（比如从3个增加到5-6个），分散State的存储压力；或者适当提高执行器的堆内内存（比如从30GB调整到35GB），但要注意Kubernetes节点的资源上限。

备注：内容来源于stack exchange，提问作者Aviral Kumar