Options
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
07-14-2022 01:50 PM
If you are looking for a graceful stop (Not to stop exactly at 5 but stop after the micro-batch that was in progress at 5 o clock instead of abruptly stopping the stream), you can try the following. The downside is if the micro batch duration is high, stream stop will be delayed.
import java.time.LocalTime
val queryStopListner = new StreamingQueryListener() {
override def onQueryStarted(queryStarted: StreamingQueryListener.QueryStartedEvent): Unit = {
}
override def onQueryTerminated(queryTerminated: StreamingQueryListener.QueryTerminatedEvent): Unit = {
}
override def onQueryProgress(queryProgress: StreamingQueryListener.QueryProgressEvent): Unit = {
val id = queryProgress.progress.id
if(LocalTime.now().isAfter(LocalTime.parse("17:00:00"))){
val currentStreamingQuery = spark.streams.get(id)
currentStreamingQuery.stop
}
}
}
//Add this query listner to the session
spark.streams.addListener(queryStopListner)