如何使用PySpark写入Kafka？遇阶段停滞与超时错误求助

我之前也碰到过类似的坑，这种卡在Stage 0然后触发Kafka元数据超时的情况，大多是网络连通性、配置参数或者版本兼容的问题，咱们一步步来排查解决：

1. Kafka Broker地址配置错误 #

这是最常见的问题，先确认你写入Kafka时的bootstrap.servers参数是否准确：

• 检查Kafka Broker的监听配置：如果是外部机器访问Kafka，Broker的server.properties里必须配置listeners=PLAINTEXT://0.0.0.0:9092（或者你的Broker公网IP），不能只绑定localhost，否则Spark根本连不上。
• 代码里要明确指定正确的Broker地址，比如：

df.write.format("kafka") \
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092") \
  .option("topic", "your_target_topic") \
  .save()

2. Spark与Kafka集群网络不通 #

Spark节点（或你运行PySpark的本地机器）和Kafka Broker之间的网络被防火墙/安全组挡住了：

• 直接在Spark节点上用telnet broker_ip 9092或者nc -zv broker_ip 9092测试连通性，连不上的话就去调整防火墙规则或者云环境的安全组，开放9092端口（或你配置的Kafka端口）。
• 如果是云环境（比如AWS、阿里云），要确保Spark所在的安全组和Kafka Broker的安全组之间有双向访问权限。

3. 目标Topic不存在且自动创建未开启 #

如果你的目标Topic还没创建，而Kafka集群关闭了自动创建Topic的功能（默认是开启的，但可能被修改），Spark会因为找不到Topic元数据而超时：

• 手动创建Topic：用Kafka命令行工具执行：

kafka-topics.sh --create --topic your_target_topic --bootstrap-server broker1:9092 --partitions 3 --replication-factor 1

• 或者检查Kafka Broker的server.properties，确保auto.create.topics.enable=true，这样Spark写入时会自动创建Topic。

4. Spark与Kafka版本不兼容 #

你用的spark-sql-kafka-0-10_2.11:2.2.0对应Kafka 0.10.x~0.11.x版本，如果你的Kafka集群是2.x及以上，很可能出现兼容性问题：

• 参照Spark官方的版本对应表，换成匹配的spark-kafka包。比如Kafka 2.4.x搭配Spark 2.4.x的话，用org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.8（根据你的Scala版本调整后缀）。

5. 元数据超时时间过短 #

默认的Kafka元数据超时是60秒，刚好和你的错误时间匹配，可以尝试调大超时参数：

df.write.format("kafka") \
  .option("kafka.bootstrap.servers", "broker1:9092") \
  .option("topic", "your_target_topic") \
  .option("kafka.metadata.max.age.ms", "120000") \  # 延长元数据过期时间
  .option("kafka.request.timeout.ms", "120000")     # 延长请求超时时间
  .save()

6. Spark集群资源不足 #

卡在Stage 0:> (0 + 8) / 9说明有9个任务只启动了8个，可能是Spark的executor资源不够，导致任务无法全部调度进而拖到超时：

• 调整Spark的资源配置，比如增大spark.executor.instances、spark.executor.cores的数值，让任务能正常启动。

快速测试建议 #

先把代码简化到最基础的版本，排除业务逻辑的干扰：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("TestKafkaWrite").getOrCreate()
# 创建最简单的测试DataFrame
test_df = spark.createDataFrame([(1, "hello kafka")], ["key", "value"])
# 转换为Kafka需要的String类型key和value
test_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
  .write.format("kafka") \
  .option("kafka.bootstrap.servers", "your_broker_ip:9092") \
  .option("topic", "test_topic") \
  .save()

如果这个测试也失败，那基本就是网络或Kafka配置的问题；如果成功，再去排查你的业务代码里的异常。

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