How do I read binary serialized Avro (Confluent Platform) from Kafka using Spark Streaming

Question

These are Avros serialized with the Confluent platform.

I would like to find a working example like this:

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

but for Spark Structured Streaming.

 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)

Did you find a working solution?

aasthetic
– aasthetic

2017-02-06 10:36:10 +00:00
Commented Feb 6, 2017 at 10:36 — aasthetic
– aasthetic, Commented Feb 6, 2017 at 10:36
@aasthetic see below

zzztimbo
– zzztimbo

2017-02-07 17:33:33 +00:00
Commented Feb 7, 2017 at 17:33 — zzztimbo
– zzztimbo, Commented Feb 7, 2017 at 17:33

Felipe Martins Melo · Accepted Answer · 2018-05-04 18:24:36Z

3

We have been working on this library which may help: ABRiS (Avro Bridge for Spark)

It provides APIs for integrating Spark to Avro in both, read and write operations (streaming and batch). It also supports Confluent Kafka and integrates with Schema Registry.

DISCLAIMER: I work for ABSA and I am the main developer behind this library.

answered May 4, 2018 at 18:24

Felipe Martins Melo

1,33311 silver badges15 bronze badges

Sign up to request clarification or add additional context in comments.

5 Comments

Pyd Over a year ago

after adding ABRiS dependencies to the POM file the method fromAvro is not there when I tried ABRiS with spark java api . Any help ?

Felipe Martins Melo Over a year ago

@Vignesh, the library is written in Scala but should be possible to be imported into a Java class. How are you trying to import it?

Pyd Over a year ago

after adding all the necessary dependencies in pom i.e(confluent and abris) I tried to import ABRiS libs as said in github.com/AbsaOSS/ABRiS Read me, but at the line

Dataset<Row> ds = sparksession.readstream().format("kafka").option("kafka.bootstrap.server","localhost:9092").option("subscribe","topicname"). method

fromavro is not there in the eclipse intellisense and there is no issue in build

Pyd Over a year ago

fromConfluentAvro is not there even after all the imports, but in scala it works fine.

Pyd Over a year ago

fromConfluentAvro method is not popping up in eclipse intelisense when we use java, but in scala it works fine. could you please shed some light on ABRiS using java

zzztimbo · Accepted Answer · 2017-02-07 17:31:18Z

2

I figured out you have to use the Confluent platform decoder if you want to read their stuff.

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

answered Feb 7, 2017 at 17:31

zzztimbo

2,3535 gold badges28 silver badges32 bronze badges

2 Comments

aasthetic Over a year ago

thanks @zzztimbo, I also registered schema for my keys, they have to be of Long type. I am unable to deserialise them in Spark Streaming. Any idea?

Bertjan Broeksema Over a year ago

It would be helpful to get a complete example of how to setup structured streaming, with a kafka topic + schema registry. It's not clear to me what to do with this decoder.

Collectives™ on Stack Overflow

How do I read binary serialized Avro (Confluent Platform) from Kafka using Spark Streaming

2 Answers 2

5 Comments

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

5 Comments

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related