Aufzüge Json Extrahieren von JSON-Objekt

Question

Ich habe diesen Code unten:

object Test { 
    def main(args: Array[String]) { 
      val sparkConf = new SparkConf().setAppName("Spark").setMaster("local[2]") 
      val sc = new SparkContext(sparkConf) 

      val ssc = new StreamingContext(sc, Seconds(3)) 
      val kafkaBrokers = Map("metadata.broker.list" -> "HostName:9092") 
      val offsetMap = Map(TopicAndPartition("topic_test", 0), 8) 
      val lines = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaBrokers, offsetMap) 

var offsetArray = Array[OffsetRange]() 
       lines.transform {rdd => 
         offsetArray = rdd.asInstanceOf[HasOffsetRanges].offsetRanges 
         rdd 
       }.map { 
         _.message() 
       }.foreachRDD {rdd => 
        /* NEW CODE */ 
       } 
       ssc.start() 
       ssc.awaitTermination() 
    } 
} 

ich den neuen Code uder der Kommentar /* NEW CODE */ hinzugefügt haben. Meine Frage ist, dass die Zeilen val eine Sequenz von RDDs enthalten, die im Grunde alle 3 Sekunden den Kafka-Server bilden. Dann greife ich die Nachricht mit der Kartenfunktion.

Aber ich bin ein wenig verwirrt darüber, was die foreachRDD Funktion tut. Ist das iterieren über alle RDD's, die in der lines DStream sind (was ist, was ich versuche zu tun)? Die Sache ist die Parse-Funktion aus der lift-json-Bibliothek akzeptiert nur eine Zeichenfolge, so dass ich über alle der RDDs durchlaufen muss und übergeben Sie diese String-Wert an die Parse-Funktion, was ich versucht habe zu tun. Aus irgendeinem Grund wird nichts ausgedruckt.

Answer 1

Wenn Sie Daten von einem bestimmten Offset lesen möchten, verwenden Sie die falsche Überladung.

Die Sie benötigen, ist dieses:

createDirectStream[K, 
        V, 
        KD <: Decoder[K], 
        VD <: Decoder[V], R] 
        (ssc: StreamingContext, 
        kafkaParams: Map[String, String], 
        fromOffsets: Map[TopicAndPartition, Long], 
        messageHandler: (MessageAndMetadata[K, V]) ⇒ R): InputDStream[R] 

Sie benötigen ein Map[TopicAndPartition, Long]:

val offsetMap = Map(TopicAndPartition("topic_test", 0), 8L) 

Und Sie brauchen eine Funktion zu übergeben, die eine MessageAndMetadata[K, V] empfängt und gibt den gewünschten Typ, zum Beispiel:

val extractKeyValue: MessageAndMetadata[String, String] => (String, String) = 
     msgAndMeta => (msgAndMeta.key(), msgAndMeta.message()) 

Und es verwenden:

KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder] 
(ssc, kafkaBrokers, offsetMap, extractKeyValue)