Scala Spark如何将现有DataFrame转换为含嵌套结构的新表

方法1：使用内置struct()函数直接构造（最常用） #

不需要手动定义StructType，Spark 内置的struct()函数可以直接将指定列打包为嵌套结构，代码最简：

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.functions.struct

val resultDF = originalDF
  .select(
    col("columnA"),
    col("columnB"),
    col("columnC"),
    // 直接将columnD、columnE打包为嵌套列columnF
    struct(
      col("columnD"),
      col("columnE")
    ).alias("columnF")
  )

如果需要自定义嵌套结构内的字段名，可给内部列加别名：

struct(
  col("columnD").alias("custom_d"),
  col("columnE").alias("custom_e")
).alias("columnF")

方法2：绑定手动定义的StructType Schema #

如果需要严格控制嵌套结构的字段约束（比如可空性、数据类型校验），可以先定义Schema再绑定数据：

第一步：定义嵌套结构Schema #

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

// 自定义columnF的结构约束
val columnFSchema = StructType(Array(
  StructField("columnD", StringType, nullable = true),
  StructField("columnE", IntegerType, nullable = false)
))

第二步：将数据绑定到Schema #

import org.apache.spark.sql.functions.{from_json, to_json}

val resultDF = originalDF
  .select(
    col("columnA"),
    col("columnB"),
    col("columnC"),
    // 先把要打包的列转成json，再用自定义Schema反序列化为嵌套结构
    from_json(
      to_json(struct(col("columnD"), col("columnE"))),
      columnFSchema
    ).alias("columnF")
  )

结果校验 #

执行printSchema()可以确认输出结构是否符合预期：

resultDF.printSchema()

预期输出结构样例：

root
 |-- columnA: [原有字段类型] (nullable = ...)
 |-- columnB: [原有字段类型] (nullable = ...)
 |-- columnC: [原有字段类型] (nullable = ...)
 |-- columnF: struct (nullable = true)
 |    |-- columnD: [指定字段类型] (nullable = ...)
 |    |-- columnE: [指定字段类型] (nullable = ...)

内容的提问来源于stack exchange，提问作者Overload Java