如何基于Case Class为Dataset扩展Schema并新增Struct列

需求回顾 #

现有Places表Schema如下：

root
|-- place_id: string (nullable = true)
|-- street_address: string (nullable = true)
|-- city: string (nullable = true)
|-- state_province: string (nullable = true)
|-- postal_code: string (nullable = true)
|-- country: string (nullable = true)
|-- neighborhood: string (nullable = true)

需生成包含subpremise字段和csm struct列的新Dataset，目标Schema：

root
|-- place_id: string (nullable = true)
|-- street_address: string (nullable = true)
|-- subpremise: string (nullable = true)
|-- city: string (nullable = true)
|-- state_province: string (nullable = true)
|-- postal_code: string (nullable = true)
|-- country: string (nullable = true)
|-- neighborhood: string (nullable = true)
|-- csm: struct (nullable = true)
|   |-- city: string (nullable = true)
|   |-- state_province: string (nullable = true)
|   |-- country: string (nullable = true)

已定义Case Class：

case class csm(
    city: Option[String] = None,
    stateProvince: Option[String] = None,
    country: Option[String] = None
)

避免withColumn手动指定列的维护问题，提供两种可行方案：

方案一：强类型Case Class映射（易维护、类型安全） #

定义包含所有目标字段的Case Class，通过map函数完成转换：

1. 定义目标Case Class

case class PlaceWithCsm(
    place_id: Option[String] = None,
    street_address: Option[String] = None,
    subpremise: Option[String] = None, // 新增字段，默认空值
    city: Option[String] = None,
    state_province: Option[String] = None,
    postal_code: Option[String] = None,
    country: Option[String] = None,
    neighborhood: Option[String] = None,
    csm: Option[csm] = None
)

2. 转换原Dataset

val placesWithCsm = places.map { row =>
    PlaceWithCsm(
        place_id = Option(row.getAs[String]("place_id")),
        street_address = Option(row.getAs[String]("street_address")),
        city = Option(row.getAs[String]("city")),
        state_province = Option(row.getAs[String]("state_province")),
        postal_code = Option(row.getAs[String]("postal_code")),
        country = Option(row.getAs[String]("country")),
        neighborhood = Option(row.getAs[String]("neighborhood")),
        // 生成csm结构，映射字段名：state_province -> stateProvince
        csm = Some(csm(
            city = Option(row.getAs[String]("city")),
            stateProvince = Option(row.getAs[String]("state_province")),
            country = Option(row.getAs[String]("country"))
        ))
        // subpremise默认None，可根据业务逻辑赋值
    )
}

优势：编译期类型检查，新增/修改字段仅需调整Case Class，维护成本低，适合长期业务迭代。

方案二：动态列处理（灵活无额外类定义） #

通过动态获取原有列，结合struct函数生成目标列，无需定义新Case Class：

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

// 自动获取原有所有列
val originalCols = places.columns.map(col)

// 构建csm struct列，处理字段名映射
val csmCol = struct(
    col("city"),
    col("state_province").alias("stateProvince"),
    col("country")
).alias("csm")

// 组合新列：原有列 + 新增subpremise（默认null） + csm列
val newCols = originalCols ++ Seq(
    lit(null).cast(StringType).alias("subpremise"),
    csmCol
)

// 生成新Dataset
val placesWithCsm = places.select(newCols: _*)

优势：无需额外类定义，自动同步原有Schema变化，适合快速迭代或Schema频繁变动的场景。

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