Blogpost: Integrating Confluent Schema Registry with Apache Spark applications

consumer.kt

fun main() {
    val config = mapOf(
        "bootstrap.servers" to "localhost:9092",
        "schema.registry.url" to "http://localhost:8081",
        "key.serializer" to "org.apache.kafka.common.serialization.StringSerializer",
        "value.serializer" to "io.confluent.kafka.serializers.KafkaAvroSerializer",
    )

    val consumer = KafkaConsumer<String, Cats>(config)
    consumer.subscribe(listOf("cats"))

    while (true) {
        val records = consumer.poll(Duration.ofSeconds(60))

        records.forEach {
            // process cat records
        }
    }
}

spark_helper.kt

fun <T> Dataset<T>.readAvro(topic: String, schemaRegistryUrl: String): Dataset<Row> {
    val fromAvroConfig = AbrisConfig
        .fromConfluentAvro()
        .downloadReaderSchemaByLatestVersion()
        .andTopicNameStrategy(topic, false)
        .usingSchemaRegistry(schemaRegistryUrl)

    return this.select(functions.from_avro(col("value"), fromAvroConfig).`as`("data"))
        .select("data.*")
}

spark_session.kt

val spark = SparkSession
      .builder()
      .appName("CatRecordConsumer")
      .orCreate

spark
    .readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "cats")
    .option("startingOffsets", "latest")
    .option("failOnDataLoss", "false")
    .load()
    .readAvro("cats", "localhost:8081")
    .writeStream()
    .format("parquet")
    .option("mergeSchema", "true")
    .option("checkpointLocation", "./_checkpoints/cats")
    .partitionBy("date")
    .start("./cats")