# Kafka Spark Streaming integration

**URL:** <https://forum.confluent.io/t/kafka-spark-streaming-integration/3606>\
**Category:** Clients\
**Created:** [9 December 2021 13:51 UTC](https://forum.confluent.io/t/kafka-spark-streaming-integration/3606 "2021-12-09T13:51:03Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Ramozanbek.Saatov](https://avatars.discourse-cdn.com/v4/letter/r/9fc348/32.png) [@Ramozanbek.Saatov](https://forum.confluent.io/u/Ramozanbek.Saatov)\
**Post date:** [9 December 2021 13:51 UTC](https://forum.confluent.io/t/kafka-spark-streaming-integration/3606/1 "2021-12-09T13:51:03Z")

</div>

Hi folks! I’m trying to consume Kafka topics by Apache Spark for structured Streaming. There’re no problems with connection and String key-values. However, I can’t find a way to use other entities with Avro deserialization.  
I want to emphasize that I’m talking about `spark-streaming-kafka-0-10` where we build streams like this

```auto
val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams),
    )

```

I’ve pushed messages with String for key and MyCustomClass for value and now I have no idea how i consume this in spark streaming. How do i deserialize the value to smth meaningful? For example, I’d like to create a Dataset[MyCustomClass] to do some transformation.  
On the internet there some examples that show how to work with spark kafka sql, but the syntax there is different. It looks smth like this

```auto
val rawTopicMessageDF = sql.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "127.0.0.1:9092")
  .option("subscribe", topicName)
  .option("startingOffsets", "earliest")
  .load()

```

Here we read some kind of dataframe already and functions like to\_avro and from\_avro are available to work with it. But what about spark-streaming-kafka way?  
The spark-streaming-kafka API gives us foreachRDD method, where we get access to records of type ConsumerRecord. I’m very confused, could anyone explain me with example code how to use spark-streaming-kafka with value deserialization?

---

<div class="post-metadata">

**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\
**Post date:** [2 March 2022 22:24 UTC](https://forum.confluent.io/t/kafka-spark-streaming-integration/3606/2 "2022-03-02T22:24:33Z")

</div>

First, with streaming approach, I don’t think you have access to `ConsumerRecord` at any point. But you could simply start with `createDirectStream[Array[Byte], Array[Byte]]` and do the deserialization on your own by instantiating `KafkaAvroDeserializer`

Structured Streaming should be preferred, however. You’d use SQL UDF functions to wrap standard Kafka (de)serializers

There are several answers here regarding the schema registry interaction - [apache kafka - Integrating Spark Structured Streaming with the Confluent Schema Registry - Stack Overflow](https://stackoverflow.com/questions/48882723/integrating-spark-structured-streaming-with-the-confluent-schema-registry)