Scala Spark中如何对含嵌套数组、结构体的Schema做全层级字母序排序

问题描述 #

实际开发中遇到的场景：

• 待处理的DataFrame原始Schema结构如下：

root
 |-- metadata2: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- attribute2: string (nullable = true)
 |    |    |-- attribute1: string (nullable = true)
 |-- metadata3: string (nullable = true)
 |-- metadata1: struct (containsNull = true)
 |    |-- attribute2: string (nullable = true)
 |    |    |-- attribute1: string (nullable = true)

• 常规操作存在的局限：
  直接调用schema.sortBy(_.name)做排序，仅会对最顶层字段完成字母排序，嵌套在Struct内部、Array承载的Struct内部的字段不会被处理，得到的Schema如下：

root
 |-- metadata1: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- attribute2: string (nullable = true)
 |    |    |-- attribute1: string (nullable = true)
 |-- metadata2: struct (containsNull = true)
 |    |-- attribute2: string (nullable = true)
 |    |    |-- attribute1: string (nullable = true)
 |-- metadata3: string (nullable = true)

• 最终预期效果：
  递归覆盖所有嵌套层级的字段排序，包括Array对应的元素Struct、嵌套Struct内部的子字段，最终得到如下结构：

root
 |-- metadata1: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- attribute1: string (nullable = true)
 |    |    |-- attribute2: string (nullable = true)
 |-- metadata2: struct (containsNull = true)
 |    |-- attribute1: string (nullable = true)
 |    |    |-- attribute2: string (nullable = true)
 |-- metadata3: string (nullable = true)

实现方案 #

核心逻辑为递归遍历Schema所有节点，针对不同字段类型做对应处理：

• 遇到StructType：先将其直属子字段按名称做字母排序，再递归处理每个子字段自身的嵌套类型，最后重新组装为新的StructType
• 遇到ArrayType：取出数组承载的元素类型，对元素类型递归完成全层级排序后，再组装回新的ArrayType
• 遇到String、Int等无嵌套结构的原子类型：直接返回原类型，无需额外处理
• 遇到MapType这类本身无序的复杂类型：可根据实际需求选择是否补充递归逻辑（默认Map的key-value本身无序，无需排序）

实现代码如下：

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

/**
 * 递归处理全层级Schema，完成所有嵌套字段的字母排序
 * @param schema 输入的原始Schema/字段类型
 * @return 排序后的新类型
 */
def sortAllNestedLevels(schema: DataType): DataType = schema match {
  // 处理StructType类型
  case struct: StructType =>
    val sortedFields = struct.fields
      .sortBy(_.name)
      .map(field => StructField(
        name = field.name,
        dataType = sortAllNestedLevels(field.dataType), // 递归处理字段的嵌套类型
        nullable = field.nullable,
        metadata = field.metadata
      ))
    StructType(sortedFields)
  // 处理ArrayType类型
  case array: ArrayType =>
    ArrayType(
      elementType = sortAllNestedLevels(array.elementType), // 递归处理数组内的元素类型
      containsNull = array.containsNull
    )
  // 其他原子类型直接返回
  case other: DataType => other
}

/**
 * 对输入DataFrame应用全层级Schema排序
 * @param df 原始DataFrame
 * @return 排序后Schema的新DataFrame
 */
def applySortedNestedSchema(df: DataFrame): DataFrame = {
  val sortedRootStruct = sortAllNestedLevels(df.schema).asInstanceOf[StructType]
  df.sqlContext.createDataFrame(df.rdd, sortedRootStruct)
}

使用说明 #

直接将原始DataFrame传入applySortedNestedSchema方法，即可得到完成全层级字段字母排序的新DataFrame；如果存在自定义的复杂嵌套类型，只需在sortAllNestedLevels方法的match分支中补充对应类型的处理逻辑即可。

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