Expand an array to columns in spark with structured streaming

user_3371468 0
Viewed 711

I have this problem:

I'm reading the data from Kafka using structured streaming, data are CSV rows. When I get the data from Kafka, I have a streaming dataframe where the CSV row is inside "value" and it's a byte sequence.

 sDF2 = sDF.selectExpr("CAST(value as string)").select( split("value",","))

using this I have a new dataframe where "value" is a string and it is the CSV row.

How can I get a new dataframe where I have parsed and split the CSV fields into dataframe columns?

Example: csv row is "abcd,123,frgh,1321" 

sDF schema, which contains the data downloaded from Kafka, is  
key, value, topic, timestamp etc... and here value is a byte sequence with no type

sDF2.schema has only a column ( named value of type string )

I like that the new dataframe is 

sDF3.col1 = abcd
sDF3.col2 = 123
sDF3.col3 = frgh ...etc

where all the columns are String.

I still can do this: 

 sDF3 = sDF2.select( sDF2.csv[0].alias("EventId").cast("string"),
 sDF2.csv[1].alias("DOEntitlementId").cast("string"),               
 sDF2.csv[3].alias("AmazonSubscriptionId").cast("string"),
 sDF2.csv[4].alias("AmazonPlanId").cast("string"),
 ... etc ...

but it looks ugly.

apache-kafka apache-spark dataframe streaming
1 Answers

I have not tried it, but something like this should work.

sDF2 = 
      sDF.selectExpr("CAST(value as string)")
       .alias("csv").select("csv.*")
       .select("split(value,',')[0] as DOEntitlementId", 
               "split(value,',')[1] as AmazonSubscriptionId", 
               "split(value,',')[2] as AmazonPlanId")
user_83816370
Related