跨机器部署Spark Streaming Kafka输出至控制台/Topic失败问题排查

问题场景 #

• 两台机器分布式部署：一台运行Kafka生产JSON数据，另一台运行Spark做聚合流式处理
• 处理后的数据无法输出到控制台或写入目标Kafka Topic，尝试多种outputMode参数均无效
• 同一脚本在单机器同时运行Spark和Kafka时完全正常
• 终止代码时触发如下异常：

ERROR:root:Exception while sending command.
Traceback (most recent call last):
File "/opt/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/clientserver.py", line 511, in send_command
answer = smart_decode(self.stream.readline()[:-1])
RuntimeError: reentrant call inside <_io.BufferedReader name=3>
处理上述异常时又触发新异常：
An error occurred while calling o76.awaitTermination

相关代码示例 #

控制台输出调试代码 #

print("Printing Schema of orders_df4: ")
orders_df4.printSchema()
# Write final result into console for debugging purpose
orders_agg_write_stream = orders_df4.writeStream \
    .trigger(processingTime='5 seconds')\
    .outputMode("update")\
    .option("truncate", "false")\
    .format("console")\
    .start()\
    .awaitTermination()

Kafka Topic写入代码 #

# Write Final RESULT To KAFKA TOPIC
orders_df4.writeStream \
    .outputMode("complete")\
    .format("kafka")\
    .option("kafka_bootsrap_servers", kafka_bootsrap_servers)\
    .option("topic", kafka_topic_write)\
    .option("checkpointLocation", "checkpoint")\
    .start()\
    .awaitAnyTermination()

排查与解决方法 #

1. 网络连通性校验

   • 测试Spark机器到Kafka机器的9092端口连通性：用telnet <kafka机器IP> 9092或nc -zv <kafka机器IP> 9092验证
   • 检查Kafka配置文件中的advertised.listeners，必须设置为Spark机器能访问的公网/跨机器内网地址，不能用localhost或仅本机可见的地址

2. 修复Py4J重入调用异常

   • 这个异常是Spark Driver和Executor的Py4J通信线程冲突导致的，避免在流处理启动前执行阻塞式IO操作（比如printSchema()可能触发线程冲突）
   • 可以把printSchema()替换为通过Spark日志查看Schema，或者将其移到流处理初始化前的独立线程中执行，不要和流处理启动代码链式调用

3. 修正代码参数与语法

   • Kafka写入代码存在参数名错误：kafka_bootsrap_servers应为kafka.bootstrap.servers（注意参数名里的点）
   • 确保checkpointLocation指向Spark机器上有读写权限的路径，若用分布式部署建议用HDFS路径
   • 拆分流处理启动和等待终止的代码，避免链式调用引发的线程阻塞：

     # 拆分写法示例
     query = orders_df4.writeStream \
         .trigger(processingTime='5 seconds')\
         .outputMode("update")\
         .option("truncate", "false")\
         .format("console")\
         .start()
     query.awaitTermination()
     

4. 适配OutputMode与聚合逻辑

   • 聚合操作的outputMode必须和逻辑匹配：
     • complete：适合全量聚合（比如全局count），每次输出所有聚合结果
     • update：仅输出有变化的聚合结果，适合增量聚合场景
   • 确认orders_df4的聚合逻辑和所选outputMode兼容，比如如果用了窗口聚合，update模式是可行的，但要确保窗口配置正确

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