我基本上是在阅读Kafka的资料，把每一条信息都传递给我的朋友 foreach 处理器（感谢jacek的页面提供了简单的示例）。
如果这真的有效，我将在 process 方法，但是，这不起作用。我相信 println 因为它在执行器上运行，所以无法将这些日志恢复到驱动程序。然而，这 insert into 一个temp表至少应该可以工作，并向我显示消息实际上已经被消费并处理到了接收器。
我错过了什么？
真的在寻找第二双眼睛来检查我的努力：

val stream = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", Util.getProperty("kafka10.broker")) 
      .option("subscribe", src_topic) 
      .load()

    val rec = stream.selectExpr("CAST(value AS STRING) as txnJson").as[(String)]

    val df = stream.selectExpr("cast (value as string) as json")

    val writer = new ForeachWriter[Row] {
      val scon = new SConConnection
      override def open(partitionId: Long, version: Long) = {
        true
      }
      override def process(value: Row) = {
        println("++++++++++++++++++++++++++++++++++++" + value.get(0))
        scon.executeUpdate("insert into rs_kafka10(miscCol) values("+value.get(0)+")")
      }
      override def close(errorOrNull: Throwable) = {
        scon.closeConnection
      }
    }

    val yy = df.writeStream
      .queryName("ForEachQuery")
      .foreach(writer)
      .outputMode("append")
      .start()

    yy.awaitTermination()