Spark Streamingを使用して、Kafkaからバイナリシリアル化されたAvro（Confluent Platform）を読むにはどうすればいいですか？

Question

これは、Confluentプラットフォームで直列化されたAvrosです。私はこのような実施例を見つけたい

：

https://github.com/seanpquig/confluent-platform-spark-streaming/blob/master/src/main/scala/example/StreamingJob.scala

が、スパーク構造化されたストリーミング用

。

kafka 
    .select("value") 
    .map { row => 

    // this gives me test == testRehydrated  
    val test = Foo("bar") 
    val testBytes = AvroWriter[Foo].toBytes(test) 
    val testRehydrated = AvroReader[Foo].fromBytes(testBytes) 


    // this yields mangled Foo data 
    val bytes = row.getAs[Array[Byte]]("value") 
    val rehydrated = AvroReader[Foo].fromBytes(bytes) 

Answer 1

私はConfluentプラットフォームデコーダを使用する必要があると考えました。

def decoder: io.confluent.kafka.serializers.KafkaAvroDecoder = { 
    val props = new Properties() 
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, getSchemaRegistryUrl()) 
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true") 
    val vProps = new kafka.utils.VerifiableProperties(props) 
    new io.confluent.kafka.serializers.KafkaAvroDecoder(vProps) 
}