spark writeStream into kafka - Difference between awaitTermination() vs awaitAnyTermination()

user_7486621 0
Viewed 946

As per official documentation, I was using below code segment to write into kafka topic, but it was not writing into kafka.

finalStream = final \
  .writeStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers",bootstrap_servers) \
  .option("topic",topic_name) \
  .option("checkpointLocation", check_point_location) \
  .start()

finalStream.awaitTermination()

But by using awaitAnyTermination() instead of awaitTermination(), writing into kafka works.

spark.streams.awaitAnyTermination()

Please suggest the reason behind this.

spark-structured-streaming pyspark apache-kafka apache-spark
1 Answers

"Difference between awaitTermination() vs awaitAnyTermination()"

Citing the comments in the Source Code

awaitTermination: "Waits for the termination of this query, either by query.stop() or by an exception. If the query has terminated with an exception, then the exception will be thrown. Otherwise, it returns whether the query has terminated or not within the timeoutMs milliseconds. If the query has terminated, then all subsequent calls to this method will either return true immediately (if the query was terminated by stop()), or throw the exception immediately (if the query has terminated with exception)."

awaitAnyTermination: "Wait until any of the queries on the associated SQLContext has terminated since the creation of the context, or since resetTerminated() was called. If any query was terminated with an exception, then the exception will be thrown. If a query has terminated, then subsequent calls to awaitAnyTermination() will either return immediately (if the query was terminated by query.stop()), or throw the exception immediately (if the query was terminated with exception). Use resetTerminated() to clear past terminations and wait for new terminations. In the case where multiple queries have terminated since resetTermination() was called, if any query has terminated with exception, then awaitAnyTermination() will throw any of the exception. For correctly documenting exceptions across multiple queries, users need to stop all of them after any of them terminates with exception, and then check the query.exception() for each query."

user_122089100
Related