Spark流处理连接Kafka报错咨询（批处理可正常运行）

代码示例 #

from pyspark.sql import SparkSession
spark = SparkSession.builder \
                    .appName('Spark') \
                    .getOrCreate()

df = spark.readStream \
    .format("kafka") \
    .option('kafka.bootstrap.servers','localhost:9092') \
    .option('subscribe', 'demo') \
    .option("failOnDataLoss","false") \
    .option('startingOffsets', 'earliest') \
    .load()

query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")\
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "demo-1") \
    .option("checkpointLocation", "/path/to/HDFS/dir") \
    .start()

query.awaitTermination()

运行命令 #

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:3.5.0 mycode.py

问题说明 #

执行到query.awaitTermination()时触发错误，但将readStream()替换为read()、writeStream()替换为write()进行批处理操作时，代码可正常运行。

可能的原因及解决方案 #

1. Checkpoint目录权限或存在性问题 #

• 若指定的HDFS目录/path/to/HDFS/dir不存在或Spark进程无读写权限，会导致流初始化失败。
• 解决：
  • 先创建目录：hdfs dfs -mkdir -p /path/to/HDFS/dir
  • 设置权限（测试环境可用，生产环境建议按需配置）：hdfs dfs -chmod 777 /path/to/HDFS/dir

2. Kafka目标主题问题 #

• 若写入的主题demo-1不存在，或Spark无该主题的写入权限，会触发错误。
• 解决：
  • 创建主题：kafka-topics.sh --create --topic demo-1 --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
  • 检查Kafka ACL配置，确保Spark账号拥有WRITE权限

3. 版本兼容性问题 #

• Spark Kafka连接器版本与Kafka服务器版本不匹配时，流处理可能报错。
• 解决：
  • 确认Kafka服务器版本，Spark 3.5.0的Kafka连接器建议搭配Kafka 2.8+版本；若版本不兼容，更换对应连接器包，比如Kafka 2.7版本可使用org.apache.spark:spark-sql-kafka-0-10_2.13:3.3.0

4. 流处理输出模式缺失 #

• 流处理需指定输出模式，默认配置可能不满足写入Kafka的要求。
• 解决：
  在writeStream中添加输出模式配置，示例如下：

  query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")\
      .writeStream \
      .format("kafka") \
      .option("kafka.bootstrap.servers", "localhost:9092") \
      .option("topic", "demo-1") \
      .option("checkpointLocation", "/path/to/HDFS/dir") \
      .outputMode("append")  # 添加输出模式配置
      .start()
  

5. 网络或Kafka服务状态问题 #

• 流处理持续连接Kafka，若网络不稳定或Kafka broker异常，会在等待终止时抛出错误。
• 解决：
  • 测试Kafka连接：telnet localhost 9092或nc -zv localhost 9092
  • 查看Kafka broker日志，确认服务正常运行

内容的提问来源于stack exchange，提问作者Thanh Tùng Lê