カフカから元のオブジェクトにバイトを変換するには？

Question

私はKafkaからデータを取り出して、デフォルトのデコーダを使用してArray[Byte]を逆シリアル化した後、(null,[B@406fa9b2)、(null,[B@21a9fe0)のように見えますが、私はこれをどのようにして元のデータにしたいのですか？

メッセージをAvro形式でシリアル化します。

Answer 1

適切なデシリアライザ、たとえば文字列またはカスタムオブジェクトを使用してバイトをデコードする必要があります。

デコードを行わないと、単純にJavaのバイト配列のテキスト表現である[B@406fa9b2が得られます。

カフカはメッセージの内容について何も知らないため、バイト配列をプロデューサからコンシューマに渡します。あなたがRDD[String]で動作するので、DStream[String]を得る上記のシリアライザで

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

：（KafkaWordCount exampleを引用）あなたはキーと値のためのシリアライザを使用する必要がストリーミングスパークで

。

バイト配列をカスタムクラスにデシリアライズしたい場合は、カスタムSerializer（これはKafka固有でSparkとは関係ありません）を記述する必要があります。

JSONを固定スキーマまたはAvro（Kafka, Spark and Avro - Part 3, Producing and consuming Avro messagesで解説されているソリューション）で使用することをお勧めします。 Structured Streamingで

---

次のようにしかし、パイプラインは見ることができる：

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