Kafkaトピックからバイナリデータを読み込みます。

Question

Kafkaのトピックから暗号化されたメッセージを読む必要があります。トピックから文字列を読み込み、私の現在のコードは次のようになります。

JavaInputDStream<ConsumerRecord<String, String>> istream1 = KafkaUtils.createDirectStream(
        jssc, 
        LocationStrategies.PreferConsistent(), 
        ConsumerStrategies.<String, String>Subscribe(Arrays.asList(topics), kafkaParams) 
       ); 

は、私が確認するために、カフカキューからバイト配列を読み取るために、このコードにどのような変更を行う必要があり、暗号化されたデータは、プロセス中に破損されません。タイプ変換の。 私はKafkaUtilsでこのAPIを見つけることができませんイム、スパークプログラミングガイドから上記のコードを取っている間： http://spark.apache.org/docs/2.0.1/api/java/org/apache/spark/streaming/kafka/KafkaUtils.html

Answer 1

あなたはカフカ接続hereの良い例を見ることができます。

しかし、あなたが望むのは、あなたのカフカトピックからByteArrayを取ることです。

この接続は次のようなJavaPairInputDStreamを作成する必要があります。応答のための

import org.apache.kafka.common.serialization.ByteArrayDeserializer 
JavaPairInputDStream<String, Array[Byte]> messages = KafkaUtils.createDirectStream(
    jssc, 
    String.class, 
    String.class, 
    StringDecoder.class, 
    ByteArrayDeserializer.class, 
    kafkaParams, 
    topicsSet 
);