Spark Structed Streaming从kafka读取嵌套的json并将其扁平化

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

val spark = SparkSession.builder()
  .appName("StructuredStreamingExample")
  .master("local[*]")
  .getOrCreate()

val schema = StructType(Seq(
  StructField("id", StringType),
  StructField("name", StringType),
  StructField("age", IntegerType),
  StructField("address", StructType(Seq(
    StructField("street", StringType),
    StructField("city", StringType),
    StructField("state", StringType)
  )))
))

val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka_server:9092")
  .option("subscribe", "topic_name")
  .load()

val jsonDF = kafkaDF.selectExpr("CAST(value AS STRING)")
  .select(from_json($"value", schema).as("data"))
  .select("data.*")

val flattenedDF = jsonDF.select(
  $"id",
  $"name",
  $"age",
  $"address.street".as("street"),
  $"address.city".as("city"),
  $"address.state".as("state")
)

val query = flattenedDF.writeStream
  .format("console")
  .outputMode("append")
  .start()

query.awaitTermination()