Wie konvertiert man Bytes von Kafka zu ihrem ursprünglichen Objekt?

Question

Ich hole Daten von Kafka und dann deserialize die Array[Byte] mit Standard-Decoder, und danach sieht meine RDD-Elemente wie (null,[B@406fa9b2), (null,[B@21a9fe0), aber ich möchte meine ursprünglichen Daten, die ein Schema haben, so wie kann ich das erreichen?

Ich serialisieren Nachrichten im Avro-Format.

Answer 1

Sie müssen die Bytes mit geeigneten Deserializern dekodieren, sagen zu Strings oder Ihrem benutzerdefinierten Objekt.

Wenn Sie nicht die Decodierung tun, erhalten Sie [B@406fa9b2 das ist einfach die Textdarstellung von Byte-Arrays in Java.

Kafka weiß nichts über den Inhalt Ihrer Nachrichten und übergibt daher Byte-Arrays von Produzenten an Konsumenten.

In Spark-Streaming Sie Serializer für Schlüssel und Werte verwenden (unter Angabe KafkaWordCount example):

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
    "org.apache.kafka.common.serialization.StringSerializer") 
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
    "org.apache.kafka.common.serialization.StringSerializer") 

Mit dem oben Serializer erhalten Sie DStream[String], so dass Sie mit RDD[String] arbeiten.

Wenn Sie jedoch Byte-Arrays direkt in Ihre benutzerdefinierte Klasse deserialisieren möchten, müssen Sie eine benutzerdefinierte Serializer schreiben (die Kafka-spezifisch ist und nichts mit Spark zu tun hat).

Was ich empfehlen würde ist die Verwendung von JSON mit einem festen Schema oder Avro (mit einer Lösung beschrieben in Kafka, Spark and Avro - Part 3, Producing and consuming Avro messages).

---

In Structured Streaming jedoch die Pipeline könnte wie folgt aussehen:

val fromKafka = spark. 
    readStream. 
    format("kafka"). 
    option("subscribe", "topic1"). 
    option("kafka.bootstrap.servers", "localhost:9092"). 
    load. 
    select('value cast "string") // <-- conversion here