Spark Structured Streaming消费Kafka遇Poll超时及任务延迟问题

spark 3.2.1  
hadoop 3.3.0  
kafka 2.5.0 
scala 2.12.12

• Kafka集群：12台VM节点
• 2个Topic，各含48个分区
• Spark配置：32个Executor，每个3核
• 消费代码配置：

val options: Map[String, String] = Map(
  "kafka.bootstrap.servers" -> "ip1:9092,ip2:9092,...,ip12:9092",
  "subscribe" -> "topic1,topic2",
  "startingOffsets" -> "latest",
  "maxOffsetsPerTrigger" -> "1000000",
  "failOnDataLoss" -> "false",
  "minPartitions" -> "48",
  )

spark.readStream.format("kafka").options(options).load()

Spark作业需对消费的数据做聚合，此前运行正常，近期出现延迟；部署两个测试消费作业也出现相同问题：

1. 单个小任务耗时极长（平均任务仅需数秒，该任务需2-3分钟甚至更久）
2. 开启spark.speculation=true及spark.speculation.quantile=0.95可触发重试，但未解决根因
3. 线程dump显示任务卡在KafkaDataConsumer.fetchData，推测在等待Kafka数据
4. Driver日志出现Consumer Poll超时告警
5. 当前Spark仅配置：

"spark.executor.heartbeatInterval": "10s"
"spark.network.timeoutInterval": "20s"
"spark.storage.blockManagerHeartbeatTimeoutMs": "20s"

1. Kafka分区与Spark任务匹配失衡：两个Topic共96个分区，当前minPartitions设为48，会导致多个Kafka分区合并到单个Spark任务中，单任务负载翻倍；同时可能存在部分Kafka分区数据分布不均，或对应Broker节点负载过高（磁盘IO、网络瓶颈），导致拉取数据等待超时。
2. Kafka Consumer核心配置缺失：未设置fetch.max.wait.ms、max.poll.interval.ms等关键参数，当分区数据写入不连续时，Consumer会等待过长时间；若Spark任务处理慢，Kafka会判定Consumer失效触发Rebalance，进一步加剧延迟。
3. Spark超时配置不合理：spark.network.timeoutInterval仅20s，远短于任务卡在数据拉取的时长，会导致Driver误判Executor状态，同时Consumer Poll超时也与网络超时、Broker响应慢直接相关。
4. Kafka Broker资源瓶颈：部分Broker节点存在CPU、磁盘IO或网络带宽过载，或Broker的replica.fetch.wait.max.ms等同步参数配置不合理，导致数据同步滞后，Spark拉取数据时等待时间过长。

1. 对齐Kafka分区与Spark任务数 #

• 修改消费代码的minPartitions为96，确保每个Kafka分区对应一个Spark任务，避免分区合并导致的负载不均；
• 检查Kafka各分区消息量，对数据量过大的分区进行拆分（需同步调整消费组的分区分配策略）。

2. 补充Kafka Consumer关键配置 #

在消费代码的options中添加以下参数：

"fetch.max.wait.ms" -> "1000", // 缩短数据拉取等待时长，避免无意义等待
"fetch.min.bytes" -> "10240", // 减少空轮询频率，降低Broker压力
"max.poll.interval.ms" -> "300000", // 延长Consumer Poll间隔，防止触发Rebalance
"session.timeout.ms" -> "30000", // 适配Spark任务时长，避免会话超时
"request.timeout.ms" -> "60000" // 延长请求超时，适配Broker响应慢的场景

3. 调整Spark超时与任务配置 #

更新Spark配置，适配数据拉取的长耗时场景：

"spark.network.timeout": "300s", // 3.2版本弃用timeoutInterval，统一用该参数设置所有网络超时
"spark.executor.heartbeatInterval": "30s",
"spark.storage.blockManagerHeartbeatTimeoutMs": "300000",
"spark.task.maxFailures": "5", // 增加任务失败重试次数
"spark.speculation.multiplier": "2" // 优化推测执行触发阈值，加快重试速度

4. 排查并优化Kafka Broker #

• 监控Kafka集群各节点的CPU、内存、磁盘IO、网络带宽，定位过载节点，进行资源扩容或数据迁移；
• 调整Broker核心参数：增加replica.fetch.max.bytes提升同步效率，降低log.flush.interval.ms减少磁盘写入延迟，根据节点CPU核数设置num.network.threads（建议等于核数）和num.io.threads（建议为核数2倍）。

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