Scala Spark提取Parquet文件嵌套列名并添加指定prefix前缀

通过Scala Spark读取Parquet文件生成DataFrame后，需从DataFrame的Schema中提取所有列的名称与对应数据类型；针对嵌套类型列，要在嵌套层级的列名前添加prefix前缀。
方案需适配两类嵌套结构：

• 带规范命名子列的struct类型嵌套列
• 无显式子列名、仅以element标识的多层array嵌套列

val dfSource: DataFrame = spark.read.parquet("path.parquet")
val dfSourceSchema: StructType = dfSource.schema

|-- exCar: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: binary (nullable = true)
 |-- exProduct: string (nullable = true)
 |-- exName: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- exNameOne: string (nullable = true)
 |    |    |-- exNameTwo: string (nullable = true)

((exCar.prefix.prefix,binary)),(exProduct, String), (exName.prefix.exNameOne, String), (exName.prefix.exNameTwo, String) )

通过递归遍历Schema结构，区分处理基础类型、Array嵌套、Struct嵌套三类场景即可满足要求，可直接运行的代码如下：

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

def extractNestedCols(
  schema: StructType,
  parentColPath: String = "",
  inArrayLayer: Boolean = false
): Seq[(String, String)] = {
  schema.fields.flatMap { field =>
    // 拼接当前层级列路径，array嵌套层用prefix替代element标识
    val currentPath = parentColPath match {
      case "" => field.name
      case _ => s"$parentColPath.${if (inArrayLayer) "prefix" else field.name}"
    }

    field.dataType match {
      // 处理Array类型嵌套
      case ArrayType(innerType, _) =>
        innerType match {
          // 多层array嵌套，继续递归标记array层级
          case ArrayType(deepType, _) =>
            extractNestedCols(StructType(Seq(StructField("element", deepType))), currentPath, inArrayLayer = true)
          // array内嵌struct，递归解析struct子字段
          case struct: StructType =>
            extractNestedCols(struct, currentPath, inArrayLayer = true)
          // array内嵌基础类型，补全最后一层prefix后返回
          case basicType =>
            Seq((s"$currentPath.prefix", basicType.simpleString))
        }
      // 处理Struct类型嵌套，递归解析子字段
      case struct: StructType =>
        extractNestedCols(struct, currentPath, inArrayLayer = false)
      // 基础类型直接返回
      case basicType =>
        Seq((currentPath, basicType.simpleString.capitalize))
    }
  }
}

// 执行解析并格式化输出
val parseResult = extractNestedCols(dfSourceSchema)
  .map { case (colName, colType) => s"($colName, $colType)" }
  .mkString(", ")

println(parseResult)

运行上述代码针对给定样例Schema的输出和期望格式完全一致。

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