PySpark与Kafka集成：批处理正常但readStream无法运行求助

Spark Structured Streaming中，writeStream.start()方法是异步启动流查询的，主线程执行该方法后不会自动阻塞等待查询运行。当代码执行完start()后，主线程没有后续阻塞逻辑，会直接结束进程，导致流查询的后台线程被终止，因此看不到任何输出就退出了。

而批处理的read()是同步操作，会阻塞主线程直到数据读取、处理和输出完成，所以能正常运行。

在流查询启动后，调用awaitTermination()方法阻塞主线程，让程序保持运行以持续处理Kafka流数据。修改后的代码如下：

from pyspark.sql import SparkSession
import logging
logging.basicConfig(level=logging.INFO)

spark = SparkSession \
    .builder \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2") \
    .appName("KafkaStreams") \
    .getOrCreate()

spark.sparkContext.setLogLevel("INFO")

# Read stream
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9094") \
    .option("subscribe", "TEST_TOPIC") \
    .option("startingOffsets", "earliest") \
    .option("kafka.sasl.mechanism", "PLAIN")\
    .option("kafka.request.timeout.ms", "60000") \
    .option("kafka.session.timeout.ms", "60000") \
    .load()

ds = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

# 启动流查询并赋值给变量
query = ds.writeStream \
    .trigger(processingTime='5 seconds') \
    .outputMode("update") \
    .format("console") \
    .option("truncate", "false") \
    .start()

# 阻塞主线程，等待流查询持续运行
query.awaitTermination()

• 测试数据流入：如果Kafka主题当前没有新数据，update输出模式不会产生任何输出。可以切换为append模式，或者手动往TEST_TOPIC发送几条测试数据验证。
• 优雅停止处理：如果需要实现优雅停止（比如响应Ctrl+C信号），可以结合信号监听和stop()方法：

  import signal
  
  def stop_streaming(signal_num, frame):
      print("Stopping streaming query gracefully...")
      query.stop()
  
  # 注册信号监听
  signal.signal(signal.SIGINT, stop_streaming)
  signal.signal(signal.SIGTERM, stop_streaming)
  
  query.awaitTermination()
  

• 固定checkpoint目录：临时checkpoint目录在查询正常退出时会被删除，建议生产环境指定固定的checkpoint目录，避免状态丢失：

  query = ds.writeStream \
      .trigger(processingTime='5 seconds') \
      .outputMode("update") \
      .format("console") \
      .option("truncate", "false") \
      .option("checkpointLocation", "/path/to/your/checkpoint") \
      .start()
  

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