Hi @Adam Rink ,
Solution 1:-
Since you're calling awaitTermination on the first query it's going to block until it completes before starting the second query.
So if you want to kick off all queries, then use:-
StreamingQueryManager.awaitAnyTermination
val query1 = df.writeStream.start()
val query2 = df.writeStream.start()
val query3....................................
spark.streams.awaitAnyTermination()
In addition to the above, by default Spark uses the FIFO scheduler. This means the first query gets all resources in the cluster while it's executing. Since you're trying to run multiple queries concurrently you should switch to the FAIR scheduler.
If you have some queries that should have more resources than the others then you can also tune the individual scheduler pools.
Solution 2:-
val query1=ds.writeSteam.{...}.start()
val query2=ds.writeSteam.{...}.start()
val query3=ds.writeSteam.{...}.start()
query3.awaitTermination()
AwaitTermination() will block your process until finished, which will never happen in a streaming app, calling it on your last query should fix your problem.
Solution 3:-
If exit on any query, use:-
spark.streams.awaitAnyTermination()
If exit on all queries:-
Option 1:-
val query1=ds.writeSteam.{...}.start()
val query2=ds.writeSteam.{...}.start()
val query3...........
query1.awaitTermination();
query2.awaitTermination();
query3................
Option 2:-
val query1=ds.writeSteam.{...}.start()
val query2=ds.writeSteam.{...}.start()
spark.streams.active.foreach(x => x.awaitTermination())
Option 3:-
while (!spark.streams.active.isEmpty) {
println("Queries currently still active: " + spark.streams.active.map(x => x.name).mkString(","))
spark.streams.awaitAnyTermination()
spark.streams.resetTerminated()
}
