Я использую структурированную потоковую передачу искры, чтобы получать данные из темы кафки и записывать данные в другой приемник кафки.
Я хочу сохранить смещение дважды - один раз при чтении из темы и пошевелить смещение. Во-вторых, при записи данных в выходной приемник и записи смещения, что возможно, указав местоположение каталога контрольной точки,
Можно ли написать смещение, израсходованное при подписке на тему.
Вы можете использовать StreamingQueryListener< /а>. Вы можете добавить слушателя в свой поток,
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 = {}
})
