Spark-Kafka集成异常：运行Spark消费脚本后Broker及生产者卡顿

问题现象 #

• 单独使用kafka-console-consumer.sh可正常读取Kafka Topic数据
• 运行Spark消费脚本后，脚本仅打印Schema就陷入卡顿，无报错信息
• 同时生产者脚本开始报超时错误并卡顿，最终Kafka Broker也出现异常卡顿

环境与已做操作 #

• 所有组件（Spark 3.5.0、Kafka 3.5.1、Scala 2.12）部署在同一EC2 Ubuntu实例上
• 已将所需JAR包放入spark/jars目录：
  • spark-sql-kafka-0-10_2.12-3.5.0.jar
  • kafka-clients-3.5.1.jar
  • spark-streaming-kafka-0-10-assembly_2.12-3.5.0.jar
  • commons-pool2-2.12.0.jar
  • spark-token-provider-kafka-0-10_2.12-3.5.0.jar
• 已尝试设置trigger和maxOffsetsPerTrigger参数，且在新实例重新部署，问题未解决

生产者脚本（producer.py，执行命令：python3 producer.py） #

from confluent_kafka import Producer
import csv
import json
import boto3
import time

kafka_config = {'bootstrap.servers': 'ec2_ip:9092'}
s3 = boto3.client('s3')
bucket_name = 'report-bucket'
csv_file = 'test_data.csv' 
topic = 'test_topic'

producer = Producer(kafka_config) 
csv_obj = s3.get_object(Bucket = bucket_name, Key = csv_file)

csv_data = csv_obj['Body'].read().decode('utf-8').splitlines()
reader = csv.DictReader(csv_data) 

print('sending data...')
for row in reader:
    data = json.dumps(row)
    print(data)
    producer.produce(topic, value=data)
    time.sleep(2)

启动消费脚本后生产者的错误信息 #

%5|1696336084.257|REQTMOUT|rdkafka#producer-1|
[thrd:localhost:9092/1]: localhost:9092/1: Timed out ProduceRequest in
flight (after 61330ms, timeout #0)
%4|1696336085.859|REQTMOUT|rdkafka#producer-1|
[thrd:localhost:9092/1]: localhost:9092/1: Timed out 1 in-flight, 0
retry-queued, 0 out-queue, 0 partially-sent requests
%3|1696336087.252|FAIL|rdkafka#producer-1| [thrd:localhost:9092/1]:
localhost:9092/1: 1 request(s) timed out: disconnect (after 81814ms in
state UP) %5|1696336114.556|REQTMOUT|rdkafka#producer-1|
[thrd:localhost:9092/1]: localhost:9092/1: Timed out ApiVersionRequest
in flight (after 9856ms, timeout #0)
%4|1696336115.042|FAIL|rdkafka#producer-1| [thrd:localhost:9092/1]:
localhost:9092/1: ApiVersionRequest failed: Local: Timed out: probably
due to broker version < 0.10 (see api.version.request configuration)
(after 11637ms in state APIVERSION_QUERY, 1 identical error(s)
suppressed) %4|1696336116.069|REQTMOUT|rdkafka#producer-1|
[thrd:localhost:9092/1]: localhost:9092/1: Timed out 1 in-flight, 0
retry-queued, 0 out-queue, 0 partially-sent requests

消费者脚本（consumer.py，执行命令：spark-submit consumer.py） #

from pyspark.sql.types import StructType, StringType, DoubleType
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json,col

spark = SparkSession.builder.appName("Kafka Consumer").getOrCreate()
kafka_broker = "ec2_ip:9092"
kafka_topic = "test_topic"

raw_df = spark.readStream.format("kafka") \
        .option("kafka.bootstrap.servers", kafka_broker) \
        .option("subscribe", kafka_topic) \
        .option("startingOffset", "earliest") \
        .load()

json_df = raw_df.selectExpr("CAST (value as STRING)")

sample_schema = StructType() \
    .add("Name", StringType(), True) \
    .add("Department", StringType(), True) \
    .add("Designation", StringType(), True) \
    .add("Salary", DoubleType(), True) \
    .add("Address", StringType(), True)


base_df = json_df.select(from_json(col("value"), sample_schema))
base_df.printSchema()
expanded_df = base_df.select("value.*")

query = expanded_df.writeStream.outputMode("append").format("console").start()
query.awaitTermination(40000)

消费脚本输出（卡顿前内容） #

Picked up _JAVA_OPTIONS: -Xmx512m
Picked up _JAVA_OPTIONS: -Xmx512m
23/10/04 06:21:25 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
root
 |-- value: string (nullable = true)

root
 |-- sample: struct (nullable = true)
 |    |-- Name: string (nullable = true)
 |    |-- Department: string (nullable = true)
 |    |-- Designation: string (nullable = true)
 |    |-- Salary: double (nullable = true)
 |    |-- Address: string(nullable = true)


23/10/04 06:21:34 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-f376c9f0-4842-4681-8fb7-14afddd1a70e. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.
23/10/04 06:21:34 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
-------------------------------------------
Batch: 0
-------------------------------------------
+--------------+-------+-----------+-----------+--------+----------+----------+--------+
|Name|Department|Designation|Salary|Address|
+--------------+-------+-----------+-----------+--------+----------+----------+--------+
+--------------+-------+-----------+-----------+--------+----------+----------+--------+

解决方案 #

1. 修复消费者脚本的Schema解析错误 #

从消费输出的Schema可见，from_json的结果默认命名为sample，但后续代码尝试用value.*读取字段，导致解析失败、无数据输出，进而引发Spark资源占用异常。修改消费者脚本中base_df的定义，给解析后的字段指定别名value：

base_df = json_df.select(from_json(col("value"), sample_schema).alias("value"))

2. 排查EC2资源瓶颈 #

Spark与Kafka同部署在一台实例上，易出现CPU、内存或磁盘IO耗尽：

• 用top/htop查看CPU、内存占用，确认是否有进程资源占满
• 用iostat检查磁盘IO使用率，磁盘繁忙会导致Kafka读写阻塞
• 临时升级EC2实例规格（增加CPU/内存）验证是否为资源问题

3. 优化生产者脚本的消息提交逻辑 #

当前生产者仅调用produce未调用flush，消息可能积压在客户端队列，加上后续卡顿引发超时。在producer.produce后添加flush：

for row in reader:
    data = json.dumps(row)
    print(data)
    producer.produce(topic, value=data)
    producer.flush()  # 立即提交消息到Broker
    time.sleep(2)

4. 调整Kafka与Spark配置 #

• Kafka Broker配置：
  • 检查server.properties的listeners，确保绑定EC2内网IP或0.0.0.0，避免仅绑定localhost导致内部通信异常
  • 提升num.network.threads和num.io.threads参数，增强Broker处理能力
• Spark配置：
  • 提交脚本时指定足够内存：spark-submit --driver-memory 2g --executor-memory 2g consumer.py
  • 增加kafka.max.poll.records控制每次拉取记录数，避免内存溢出：

    raw_df = spark.readStream.format("kafka") \
            .option("kafka.bootstrap.servers", kafka_broker) \
            .option("subscribe", kafka_topic) \
            .option("startingOffset", "earliest") \
            .option("kafka.max.poll.records", "1000") \
            .load()
    

5. 检查网络与端口 #

• 确认EC2安全组允许9092端口的内部通信（即使同一实例，也要确保端口未被防火墙拦截）
• 检查Kafka的advertised.listeners配置，确保与生产者、消费者使用的地址一致

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