spark.streams.addListener(new StreamingQueryListener() {
override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
// insert code here to log the offsets in addition to Spark's checkpoint
}
override def onQueryProgress(event: QueryProgressEvent): Unit = {}
override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {}
})
1条答案
按热度按时间4zcjmb1e1#
您可以使用streamingquerylistener。您可以通过以下方式将侦听器添加到流中