Czy możliwe jest uzyskanie określonego przesunięcia komunikatu w Kafce + SparkStreaming?

Question

Próbuję uzyskać i zapisać przesunięcie dla konkretnej wiadomości w Kafce, używając strumienia bezpośredniego Spark. Przeglądanie dokumentacji Spark to prosty sposób na uzyskanie przesunięć zakresu dla każdej partycji, ale potrzebuję zapisać offset początkowy dla każdej wiadomości tematu po pełnym skanowaniu kolejki.

Answer 1

Tak, można użyć wersji MessageAndMetadata w wersji createDirectStream, która umożliwia dostęp do message metadata.

Możesz znaleźć przykład tutaj, który zwraca Dstream z 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) 
    }) 

W powyższym przykładzie tuple3._1 będzie miał topic, tuple3._2 będzie miał offset i tuple3._3 będzie miał message.

Mam nadzieję, że to pomoże!