Queries with streaming sources must be executed with writeStream.start();

by

You are branching the query plan: from the same ds1 you are trying to:

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

But you are only calling .start() on the second branch, leaving the other dangling without a termination, which in turn throws the exception you are getting back.

The solution is to start both branches and await termination.

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

ds1.printSchema()
query1.awaitTermination()
query2.awaitTermination()

More Related Contents:

  1. Spark DataFrame: does groupBy after orderBy maintain that order?
  2. How to calculate the size of dataframe in bytes in Spark?
  3. How to convert rdd object to dataframe in spark
  4. How to define and use a User-Defined Aggregate Function in Spark SQL?
  5. Automatically and Elegantly flatten DataFrame in Spark SQL
  6. Spark extracting values from a Row
  7. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  8. Provide schema while reading csv file as a dataframe
  9. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  10. Processing multiple files as independent RDD’s in parallel
  11. How to flatmap a nested Dataframe in Spark
  12. MatchError while accessing vector column in Spark 2.0
  13. How to save DataFrame directly to Hive?
  14. Spark / Scala: forward fill with last observation
  15. About how to add a new column to an existing DataFrame with random values in Scala
  16. Joining Spark dataframes on the key
  17. How to use Column.isin with list?
  18. How to write unit tests in Spark 2.0+?
  19. How to compare two dataframe and print columns that are different in scala
  20. Apache Spark: Get number of records per partition
  21. Renaming column names of a DataFrame in Spark Scala
  22. Derive multiple columns from a single column in a Spark DataFrame
  23. How to save a spark DataFrame as csv on disk?
  24. What is going wrong with `unionAll` of Spark `DataFrame`?
  25. How to sort by column in descending order in Spark SQL?
  26. Exploding nested Struct in Spark dataframe
  27. How to get keys and values from MapType column in SparkSQL DataFrame
  28. How to create a Dataset of Maps?
  29. Spark DataFrames when udf functions do not accept large enough input variables
  30. Filter spark DataFrame on string contains

Leave a Comment