Spark Structured Streaming问题：StreamingQuery.awaitTermination()完成Kafka数据写入后无法退出

我太懂你碰到的这个糟心问题了——数据明明已经成功写到Kafka里，控制台消费者都能看到消息，可程序就是卡在awaitTermination()那里死活不退出，完全僵住，对吧？

问题根源 #

你用readStream读取的是静态文件夹里的JSON文件，但Structured Streaming的默认设计是持续监听数据源的新数据。哪怕文件夹里的所有文件都处理完了，流查询还是会一直处于“等待新文件出现”的状态，不会自动终止，所以awaitTermination()会一直阻塞，程序自然无法退出。

解决方案 #

根据你“一次性处理数据后终止程序”的需求，有两种最直接的解决方式：

1. 使用Trigger.Once()（推荐流处理场景）

这是Spark专门为“批处理式流任务”设计的触发器，它会告诉Spark：只处理当前可用的所有数据，处理完成后自动停止查询。修改你的代码如下：

import org.apache.spark.sql.streaming.Trigger // 需要导入这个类

def writeToKafka() = {
  val carsDF = spark.readStream
    .schema(carsSchema)
    .json("src/main/resources/data/cars")

  val carsKafkaDF = carsDF.selectExpr("upper(Name) as key", "Name as value")

  val query = carsKafkaDF.writeStream.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("topic", "rockthejvm")
    .option("checkpointLocation", "checkpoints_demo")
    .trigger(Trigger.Once()) // 关键：触发一次处理后自动停止
    .start()

  query.awaitTermination() // 现在处理完所有数据后会自动返回，程序正常退出
}

加上Trigger.Once()后，流查询会在处理完文件夹里的所有JSON文件后自动停止运行，awaitTermination()也就会正常返回，程序自然终止。

2. 直接用批处理（更适合静态数据场景）

如果你的需求只是一次性读取静态文件并写入Kafka，其实完全没必要用流处理，直接用Spark的批处理API更简单高效：

def writeToKafkaBatch() = {
  val carsDF = spark.read
    .schema(carsSchema)
    .json("src/main/resources/data/cars")

  val carsKafkaDF = carsDF.selectExpr("upper(Name) as key", "Name as value")

  carsKafkaDF.write.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("topic", "rockthejvm")
    .save()
}

这种方式没有流查询的持续监听逻辑，处理完数据后程序直接退出，完全不会有冻结的问题。

额外说明 #

如果以后你需要处理持续生成的流数据（比如文件夹会不断新增文件），那默认的流查询行为是合理的——它会一直运行等待新数据。但对于一次性的静态数据处理，上述两种方式都能完美解决你的冻结问题。

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