How to parse a json string column in pyspark's DataStreamReader and create a Data Frame

Question

I am reading messages from a kafka topic

messageDFRaw = spark.readStream\
                    .format("kafka")\
                    .option("kafka.bootstrap.servers", "localhost:9092")\
                    .option("subscribe", "test-message")\
                    .load()

messageDF = messageDFRaw.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING) as dict")

When I print the data frame from the above query I get the below console output.

|key|dict|
|#badbunny |{"channel": "#badbunny", "username": "mgat22", "message": "cool"}|

I tried following the accepted answer in How to read records in JSON format from Kafka using Structured Streaming?

struct = StructType([
    StructField("channel", StringType()),
    StructField("username", StringType()),
    StructField("message", StringType()),
])

messageDFRaw.select(from_json("CAST(value AS STRING)", struct))

but, I get Expected type 'StructField', got 'StructType' instead in from_json()

pulsar · Accepted Answer · 2019-02-16 23:08:17Z

1

I ignored the warning Expected type 'StructField', got 'StructType' instead in from_json().

However, I had to cast the value from kafka message initially and then parse the json schema later.

messageDF = messageDFRaw.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

messageParsedDF = messageDF.select(from_json("value", struct_schema).alias("message"))

messageFlattenedDF = messageParsedDF.selectExpr("value.channel", "value.username", "value.message")

answered Feb 16, 2019 at 23:08

pulsar

1412 silver badges14 bronze badges

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

Collectives™ on Stack Overflow

How to parse a json string column in pyspark's DataStreamReader and create a Data Frame

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related