How to read records in JSON format from Kafka using Structured Streaming?

by

From the Spark perspective value is just a byte sequence. It has no knowledge about the serialization format or content. To be able to extract the filed you have to parse it first.

If data is serialized as a JSON string you have two options. You can cast value to StringType and use from_json and provide a schema:

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.from_json

val schema: StructType = StructType(Seq(
  StructField("column1", ???),
  StructField("column2", ???)
))

rawKafkaDF.select(from_json($"value".cast(StringType), schema))

or cast to StringType, extract fields by path using get_json_object:

import org.apache.spark.sql.functions.get_json_object

val columns: Seq[String] = ???

val exprs = columns.map(c => get_json_object($"value", s"$$.$c"))

rawKafkaDF.select(exprs: _*)

and cast later to the desired types.

More Related Contents:

  1. Spark – load CSV file as DataFrame?
  2. How to define partitioning of DataFrame?
  3. Encoder error while trying to map dataframe row to updated row
  4. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  5. How to define schema for custom type in Spark SQL?
  6. How can I change column types in Spark SQL’s DataFrame?
  7. Spark unionAll multiple dataframes
  8. How can I pass extra parameters to UDFs in Spark SQL?
  9. SparkSQL: How to deal with null values in user defined function?
  10. How to split a dataframe into dataframes with same column values?
  11. How to aggregate values into collection after groupBy?
  12. Dropping a nested column from Spark DataFrame
  13. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  14. Spark DataFrame: does groupBy after orderBy maintain that order?
  15. Filling gaps in timeseries Spark
  16. Spark UDAF with ArrayType as bufferSchema performance issues
  17. Apache Spark how to append new column from list/array to Spark dataframe
  18. How to use COGROUP for large datasets
  19. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  20. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  21. Create new Dataframe with empty/null field values
  22. Perform a typed join in Scala with Spark Datasets
  23. DataFrame-ified zipWithIndex
  24. Spark: Add column to dataframe conditionally
  25. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  26. Calculate Cosine Similarity Spark Dataframe
  27. Why is join not possible after show operator?
  28. How to create DataFrame from Scala’s List of Iterables?
  29. How to get Kafka offsets for structured query for manual and reliable offset management?
  30. How to create a Dataset of Maps?

Leave a Comment