0

Is there some easy way how to save a spark structured streaming dataframe into kafka with Confluent Schema registry? Spark version is 3.2.0, Scala 2.12

I managed to read data from Kafka with Confluent schema registry with a bit of an ugly code:

  val schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistry, 128)
  val kafkaAvroDeserializer = new AvroDeserializer(schemaRegistryClient)
  val deserializer = kafkaAvroDeserializer
}

class AvroDeserializer extends AbstractKafkaAvroDeserializer {
  def this(client: SchemaRegistryClient) {
    this()
    this.schemaRegistry = client
  }

  override def deserialize(bytes: Array[Byte]): String = {
    val genericRecord = super.deserialize(bytes).asInstanceOf[GenericRecord]
    genericRecord.toString
  }
}

spark.udf.register("deserialize", (bytes: Array[Byte]) =>
  DeserializerWrapper.deserializer.deserialize(bytes))```

Now I would like to write the data to another Kafka topic - is there a simple way?

1 Answer 1

1

You'd need to use similarly ugly code that uses a serializer UDF over a Struct column (or primitive type).

There's libraries that can help with making it less ugly - https://github.com/AbsaOSS/ABRiS

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.