È possibile ottenere uno specifico offset di messaggio in Kafka + SparkStreaming?

Question

Sto cercando di ottenere e memorizzare l'offset per un messaggio specifico in Kafka utilizzando Spark Direct Stream. Guardare la documentazione di Spark è semplice per ottenere gli offset di gamma per ogni partizione, ma quello di cui ho bisogno è di memorizzare l'offset iniziale per ogni messaggio di un argomento dopo una scansione completa della coda.

Answer 1

Sì, è possibile utilizzare la versione MessageAndMetadata di createDirectStream che consente di accedere a message metadata.

È possibile trovare un esempio qui che restituisce Dstream di tuple3.

val ssc = new StreamingContext(sparkConf, Seconds(10)) 

val kafkaParams = Map[String, String]("metadata.broker.list" -> (kafkaBroker)) 
var fromOffsets = Map[TopicAndPartition, Long]() 
val topicAndPartition: TopicAndPartition = new TopicAndPartition(kafkaTopic.trim, 0) 
val topicAndPartition1: TopicAndPartition = new TopicAndPartition(kafkaTopic1.trim, 0) 
fromOffsets += (topicAndPartition -> inputOffset) 
fromOffsets += (topicAndPartition1 -> inputOffset1) 

val messagesDStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, Tuple3[String, Long, String]](ssc, kafkaParams, fromOffsets, (mmd: MessageAndMetadata[String, String]) => { 
    (mmd.topic ,mmd.offset, mmd.message().toString) 
    }) 

Nel precedente esempio tuple3._1 avranno topic, tuple3._2 avrà offset e tuple3._3 avrà message.

Spero che questo aiuti!