Reading Kafka Connect JSONConverter messages with schema using Spark Structured Streaming

Question

I am trying to read message from Kafka Topic. Message are in below format (sample format):

{"schema":{"type":"struct","name":"emp_table","fields":[{"field":"emp_id","type":"string"},{"field":"emp_name","type":"String"},{"field":"city","type":"string"},{"field":"emp_sal","type":"string"},{"field":"manager_name","type":"string"}]},"payload":{"emp_id":"1","emp_name":"abc","city":"NYK","emp_sal":"100000","manager_name":"xyz"}}

Also, please note topic has message from different tables and not just 1 table.

What I am trying to achieve is to read above message from Kafka Topic using Spark Structured Streaming and create a dataframe with column names ad its value both coming from JSON message itself.

I don't want to explicitly define a schema using case class or StructType.

I tried this:

val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", brokers).option("subscribe", "topic1").option("startingOffsets", "earliest").load()

val y=df.select(get_json_object(($"value"), "$.payload").alias("payload")

When I view Y (which is a dataframe), it comes as 1 column with value under payload as JSON in that column.

How to get individual column in a dataframe? I am not achieve this.

(Again reiterating I cannot use a generic case class or StructType for schema part as messages coming through Kafka message are from different tables so I want a more of dynamic Schema created from JSON itself on the run.)

Umh, you could probably access to single values via dot notation, payload.schema.type should return "struct" as value — Tizianoreica
– Tizianoreica, Commented Jan 7, 2019 at 8:25
I think you need to cast the value to a string (from a byte array) before you can use get_json_object — OneCricketeer
– OneCricketeer, Commented Jan 13, 2019 at 17:51

OneCricketeer · Accepted Answer · 2019-09-17 16:41:54Z

1

Option 1: Change the Kafka Connect source to set value.converter.schemas.enable=false. This will only give you the (unwrapped payload to begin with), then you can skip to below post.

Otherwise, after you strip the Connect schema, you would need to use from_json() to apply a schema

val y = df.select(get_json_object($"value", "$.payload").alias("payload"))
val z = df.select(from_json($"payload", schema))

All your fields are strings, so would look like

val schema: StructType = StructType(Seq(
  StructField("emp_id", StringType()),
  StructField("emp_name", StringType()),
  StructField("city", StringType()),
  StructField("emp_sal", StringType()),
  StructField("manager_name", StringType())
))

answered Jun 5, 2019 at 3:39

OneCricketeer

193k20 gold badges146 silver badges276 bronze badges

Sign up to request clarification or add additional context in comments.

6 Comments

soMuchToLearnAndShare Over a year ago

Hi @OneCriketeer, if my payload is like this ``` "payload": { "data": { "nameField": "Myname", "AnotherField": [ { "value": { "unitField": "MyUnit", "valueField": "27" } } ], "timestampField": "2020-08-01T18:00:00" } } ``` , would i need to define some kind of nested struct?

soMuchToLearnAndShare Over a year ago

hmm. find something here: stackoverflow.com/questions/57079343/…

OneCricketeer Over a year ago

@Minnie AnotherField would need to be an Array type, yes

soMuchToLearnAndShare Over a year ago

I am stuck and I will post my question below to keep the topic concentrated.

OneCricketeer Over a year ago

@Minnie Please create a new post rather than ask a question after an accepted answer

|

Collectives™ on Stack Overflow

Reading Kafka Connect JSONConverter messages with schema using Spark Structured Streaming

1 Answer 1

6 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

6 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related