Kafka + SparkStreamingで特定のメッセージオフセットを取得することはできますか？

Question

Spark Direct Streamを使用して、カフカの特定のメッセージのオフセットを取得して保存しようとしています。 Sparkのドキュメントは、各パーティションの範囲オフセットを取得するのは簡単ですが、必要なのはキューのフルスキャン後にトピックの各メッセージの開始オフセットを格納することです。

Answer 1

はい、MessageAndMetadataバージョンcreateDirectStreamを使用すると、message metadataにアクセスできます。

tuple3のDstreamを返す例があります。上記の例で

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) 
    }) 

tuple3._1はtopic、tuple3._2はoffsetとtuple3._3はmessageを有している必要がありますがあります。

希望すると便利です。