PySpark中Kafka Spark Streaming无法读取数据求助

• 检查Kafka主题是否有新数据
  流处理默认消费新产生的消息，如果METRICS主题当前没有新数据写入，流会一直处于等待状态，不会输出任何内容。而批处理是读取主题中已有的全部数据，所以能正常返回结果。可以用Kafka命令行工具验证：

  kafka-console-consumer.sh --bootstrap-server b1,b2,b3 --topic METRICS --from-beginning
  

  如果没有新消息，尝试往主题中写入测试数据，再运行流处理代码。

• 修复代码语法错误
  你提供的控制台输出代码中，queryName("tarana")后缺少了连接符.，这会导致语法错误，程序无法正常启动。修正后的代码：

  query = df.selectExpr("CAST(value AS STRING)") \
      .writeStream \
      .queryName("tarana") \
      .format("console") \
      .outputMode("append") \
      .start()
  
  query.awaitTermination()
  

• 配置起始偏移量消费历史数据
  默认情况下，流处理从主题的最新偏移量开始消费。如果想验证流处理是否能正常运行，可以配置从最早偏移量开始，消费主题中已有的数据：

  df = spark \
      .readStream \
      .format("kafka") \
      .option("kafka.bootstrap.servers", "b1,b2,b3") \
      .option("subscribe", "METRICS") \
      .option("includeHeaders", "true") \
      .option("startingOffsets", "earliest")  # 新增此配置
      .load()
  

• 查看Databricks流处理日志
  在Databricks集群的Logs页面，查看流处理任务的详细日志，排查是否存在权限不足、序列化错误或其他隐藏问题。批处理能运行不代表流处理的权限完全一致，比如流处理用户可能缺少Kafka主题的持续消费权限。

• 验证Spark与Kafka版本兼容性
  不同Spark版本对Kafka的版本有兼容性要求（例如Spark 3.0+兼容Kafka 2.0及以上版本）。如果版本不匹配，可能导致流处理卡住。可以在Databricks集群的Spark配置中查看版本，对比Kafka集群版本是否符合官方兼容要求。

• 简化输出逻辑验证
  先避免复杂的字段转换，直接输出Kafka原始字段（如offset、key）来验证流是否正常工作：

  query = df.select("offset", "key", "value") \
      .writeStream \
      .format("console") \
      .outputMode("append") \
      .start()
  query.awaitTermination()
  

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