Spark Structured Streaming：Console Sink无法正常输出问题排查

首先，你遇到的这个现象确实和Spark Structured Streaming的状态管理、Schema匹配以及Console Sink的特性都有关系，咱们一步步拆解：

1. Case Class与DataFrame Schema不匹配（最可能的直接原因） #

看你的代码，kafkaStreamingDF通过selectExpr获取了三个字段：value、timestamp、topic，但你定义的case class record只有value和topic两个属性。当你调用as[record]进行类型转换时，Spark会尝试自动映射字段，但因为timestamp没有对应的case class成员，这个转换过程会静默丢弃timestamp字段，更关键的是，这种不匹配可能导致部分数据无法正确解析为record实例——而自定义ForeachWriter因为只用到了存在的value和topic，所以能正常打印；但Console Sink对Schema一致性的校验更严格，会导致数据无法正常输出。

修复方案：
要么修改case class包含timestamp字段：

case class record(value: String, topic: String, timestamp: String)

要么在selectExpr里只保留case class需要的字段，避免多余字段干扰：

val kafkaStreamingDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "...")
  .option("subscribe", "...")
  .option("failOnDataLoss", "false")
  .option("startingOffsets","earliest")
  .load()
  .selectExpr("CAST(value as STRING)", "CAST(topic as STRING)") // 只保留case class对应的字段

2. Console Sink的输出配置问题 #

Console Sink默认的触发逻辑是微批完成即输出，如果你的数据流速度很快或者数据量极小，可能微批瞬间完成，输出被控制台日志淹没；另外默认会截断长字符串，也可能导致你误以为没有输出。

优化配置：
添加触发间隔和关闭截断的配置，确保输出可见：

import org.apache.spark.sql.streaming.Trigger

val query = sdvDF
  .writeStream
  .format("console")
  .trigger(Trigger.ProcessingTime("5 seconds")) // 每5秒触发一次微批，便于观察输出
  .option("truncate", "false") // 不截断输出内容，完整显示数据
  .start()
  .awaitTermination()

3. Checkpoint状态的影响（你猜测的点确实存在） #

Spark Structured Streaming的Checkpoint机制会记录作业的消费偏移量和状态，如果你之前运行过相同的流作业（哪怕用的是不同Sink），并且没有指定独立的Checkpoint目录，Spark会复用之前的状态——比如之前用Foreach已经消费过所有数据，Console Sink会从上次结束的偏移量开始，自然没有新数据输出。

解决方法：
给Console Sink指定独立的Checkpoint目录，确保它使用自己的状态：

val query = sdvDF
  .writeStream
  .format("console")
  .option("checkpointLocation", "/tmp/console-spark-checkpoint") // 独立的临时Checkpoint目录
  .start()
  .awaitTermination()

如果是本地运行，也可以重启IntelliJ和Spark进程，清除本地的临时状态缓存。

4. 验证Filter条件 #

虽然你的Foreach能正常输出，但可以临时注释掉.filter($"value".isNotNull)，测试Console Sink是否能输出数据——如果此时能输出，说明你的数据流中存在大量value为null的消息，Filter过滤后没有剩余数据。

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