PySpark Streaming作业在Kafka集群恢复后卡住不消费该如何排查

问题根因 #

1. Spark 2.4.5内置Kafka客户端默认配置容错能力不足
   • 默认kafka.consumer.connections.max.idle.ms参数值为540秒，Kafka故障时间超过该阈值后消费者空闲连接会被关闭，而Spark 2.4.x版本的Structured Streaming Kafka源不会自动重建连接
   • 默认重试次数、请求超时、心跳相关参数配置不合理，Kafka集群恢复后，原消费者实例已经被集群踢出消费组，Spark端无法自动触发重平衡拉取新数据
2. 版本存在已知bug：Spark 2.4.5存在社区已确认的Structured Streaming Kafka源在Broker故障后无法自动恢复的问题，对应ISSUE SPARK-27819、SPARK-25117，相关问题在Spark 3.0+版本才完全修复
3. 无报错日志的原因：Kafka消费者的连接重试、等待逻辑在客户端内部静默执行，重试耗尽后线程进入阻塞状态，没有抛出未捕获异常到Spark任务调度层，因此Driver和Executor都不会打印错误日志

解决方案 #

临时修复（无需升级Spark版本，快速生效） #

在原有Kafka读取配置中增加以下消费者容错参数即可：

inputDF = spark \
                .readStream \
                .format("kafka") \
                .option("kafka.bootstrap.servers", Kakfa_broker_ip:port) \
                .option("subscribe", "kafka_topic") \
                # 新增容错配置
                .option("kafka.consumer.connections.max.idle.ms", "86400000") # 连接空闲超时设为1天，避免长时间无数据断开
                .option("kafka.consumer.request.timeout.ms", "300000") # 请求超时设为5分钟
                .option("kafka.consumer.retry.backoff.ms", "1000") # 重试间隔1秒
                .option("kafka.consumer.retries", "1000") # 最大重试次数1000次
                .option("kafka.consumer.max.poll.interval.ms", "300000") # 最大拉取间隔5分钟，避免消费慢被踢出组
                .option("kafka.consumer.session.timeout.ms", "60000") # 会话超时60秒
                .option("kafka.consumer.heartbeat.interval.ms", "10000") # 心跳间隔10秒
                .option("failOnDataLoss", "false") # 非严格数据一致性场景下允许继续消费，避免任务挂死
                .load() \
                .selectExpr("CAST(value AS STRING)") \
                .writeStream \
                .trigger(processingTime='30 seconds') \
                .option("checkpointLocation", "/file-storage/Checkpoints")\
                .foreachBatch(processKafkaInput) \
                .outputMode("append") \
                .start()

调整配置后重启任务，Kafka集群故障恢复后即可自动重连继续消费。

兜底方案 #

在Kubernetes侧配置任务健康检查，通过Spark UI的流式任务指标（最近一次批处理完成时间、单位时间处理记录数）判断任务状态，超过阈值（例如30分钟没有新的批处理执行）自动重启任务。

永久修复 #

升级Spark版本到3.0及以上，该版本已经彻底修复了Kafka源故障后无法自动重连的相关bug，配合上述参数配置可实现更高的容错性。

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