How to use from_json with Kafka connect 0.10 and Spark Structured Streaming?

I was trying to reproduce the example from [Databricks][1] and apply it to the new connector to Kafka and spark structured streaming however I cannot parse the JSON correctly using the out-of-the-box methods in Spark…

note: the topic is written into Kafka in JSON format.

val ds1 = spark
          .option("kafka.bootstrap.servers", IP + ":9092")
          .option("zookeeper.connect", IP + ":2181")
          .option("subscribe", TOPIC)
          .option("startingOffsets", "earliest")
          .option("max.poll.records", 10)
          .option("failOnDataLoss", false)

The following code won’t work, I believe that’s because the column json is a string and does not match the method from_json signature…

    val df =$"value" cast "string" as "json")
                .select(from_json("json") as "data")

Any tips?

[UPDATE] Example working:


First you need to define the schema for your JSON message. For example

val schema = new StructType()

Now you can use this schema in from_json method like below.

val df =$"value" cast "string" as "json")
            .select(from_json($"json", schema) as "data")

Source : Link , Question Author : carlos rodrigues , Answer Author : Jacek Laskowski

Leave a Comment