initialisieren createDirectStream mit Message

Question

Unten ist der bestehende Code, der i DSTREAM [Zeichenfolge] von Kafka Spark-Streaming lesen bin mit erstellen:

def createDirectStream[K, V, KD <: Decoder[K], VD <: Decoder[V]](ssc: StreamingContext, 
    kafkaParams: Map[String, String], topics: Set[String])(implicit arg0: ClassTag[K], 
    arg1: ClassTag[V], arg2: ClassTag[KD], arg3: ClassTag[VD]): InputDStream[(K, V)] 

zu

:

var messages: DStream[String] = null 
val topicsSet = kafkaTopicsName.split(",").toSet 
messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](streamingContext, kafkaParams, topicsSet).map(_._2) 
message.foreachRDD { rdd => 
. . . . . 
} 
streamingContext.start() 
streamingContext.awaitTermination() 

ich wechseln wollte,

def createDirectStream[K, V, KD <: Decoder[K], VD <: Decoder[V], R] 
    (jssc: JavaStreamingContext, keyClass: Class[K], valueClass: Class[V], 
    keyDecoderClass: Class[KD], valueDecoderClass: Class[VD], recordClass: Class[R], 
    kafkaParams: Map[String, String], fromOffsets: Map[TopicAndPartition, Long], 
    messageHandler: Function[MessageAndMetadata[K, V], R]): JavaInputDStream[R] 

So dass ich den Offset von Funken Verbraucher auf eigene Faust verwalten kann. Ich habe herausgefunden, wie Offset-Bereiche von Zookeeper zugreifen und Offset-Bereiche zu createDirectStream zuweisen

Aber ich bin nicht sicher darüber, wie übergeben MessageHDler und Klasse R. Was genau diese beiden Parameter tut? Wirklich gut, wenn ich ein Beispielcode-Snippet in scala für den obigen Konstruktor bekommen kann.

Answer 1

Sie können so etwas verwenden.

KafkaUtils.createDirectStream (sparkStreamContext, String.class, String.class, StringDecoder.class, StringDecoder.class, String.class, kafkaParams, fromOffsets, (Funktion, String>) MessageAndMetadata :: Nachricht);

Dies wird zurückkehren rechnete eine JavaInputDStream aus Ihrer Nachricht, basierend auf dem fromOffsets

Answer 2

ich eine Lösung aus:

val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
    streamingContext, kafkaParams, topicMap, (mmd: MessageAndMetadata[String, String]) => mmd.message())