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?