Abfragen mit Streaming-Quellen müssen mit writeStream.start() ausgeführt werden;

Question

Ich versuche, die Nachrichten von Kafka (Version 10) in Funke zu lesen und zu versuchen, es zu drucken.

 import spark.implicits._ 

     val spark = SparkSession 
       .builder 
       .appName("StructuredNetworkWordCount") 
       .config("spark.master", "local") 
       .getOrCreate() 

      val ds1 = spark.readStream.format("kafka") 
    .option("kafka.bootstrap.servers", "localhost:9092") 
.option("subscribe", "topicA") .load() 
      ds1.collect.foreach(println) 
     ds1.writeStream 
      .format("console") 
      .start() 
      ds1.printSchema() 

eine Fehlerausnahme in thread "main" org.apache.spark.sql.AnalysisException erhalten: Abfragen mit Streaming Quellen müssen mit writeStream.start() ausgeführt werden ;;

Answer 1

Sie verzweigen den Abfrageplan: aus dem gleichen ds1 Sie versuchen, zu:

• ds1.collect.foreach(...)
• ds1.writeStream.format(...){...}

Aber Sie sind nur .start() auf dem zweiten Zweig aufrufen, so dass die andere baumeln ohne eine Kündigung, die wiederum die Ausnahme wirft, die Sie zurückbekommen.

Answer 2

Ich konfrontiert dasselbe Problem.I behobenes Problem mit dem folgenden Code. Es könnte hilfreich sein, das Problem zu beheben.

val df = session 
    .readStream 
    .format("kafka") 
    .option("kafka.bootstrap.servers", brokers) 
    .option("subscribe", "streamTest2") 
    .load(); 

    val query = df.writeStream 
    .outputMode("append") 
    .format("console") 
    .start() 
query.awaitTermination() 

Answer 3

wenn es um die Fehlermeldung zu lesen

org.apache.spark.sql.AnalysisException: Abfragen mit Streaming-Quellen muss mit writeStream.start ausgeführt werden() ;;

ich fand diese article, das erklärt es gut und bietet eine andere procroche. Ich werde es selbst ausprobieren und die Ergebnisse später veröffentlichen, wenn es für mich funktioniert.