Scala处理嵌套JSON：将DataFrame各级列名的-替换为_

问题背景 #

我用Scala处理嵌套JSON格式的DataFrame，需要把所有层级列名中的“-”替换为“_”。但目前写的代码只能替换第一层级的列名，求一个能适配任意嵌套结构的动态解决方案。

目标JSON Schema（原始） #

|-- a-type: struct (nullable = true)
 |    |-- x-Type: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- part: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- x-Type: array (nullable = true)
 |    |    |    |    |-- element: string (containsNull = true)
 |    |    |    |-- Length: long (nullable = true)
 |    |    |    |-- Order: long (nullable = true)
 |    |    |    |-- y-Name: string (nullable = true)
 |    |    |    |-- Payload-Text: string (nullable = true)
 |    |-- Date: string (nullable = true)

现有代码（仅支持第一层级替换） #

scJsonDF.columns.foreach { col =>
    println(col + " after column replace " + col.replaceAll("-", "_"))
    scJsonDFCorrectedCols = scJsonDFCorrectedCols.withColumnRenamed(col, col.replaceAll("-", "_")
    )
}

动态解决方案 #

要处理所有嵌套层级，必须通过递归遍历Schema结构，针对Struct、Array等复杂类型分别处理：

核心递归处理函数 #

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

def renameNestedColumns(schema: StructType, parentPath: String = ""): Array[Column] = {
  schema.fields.flatMap { field =>
    // 替换当前字段名的“-”为“_”
    val newFieldName = field.name.replaceAll("-", "_")
    // 构造当前字段的完整路径（用于定位原始字段）
    val originalPath = if (parentPath.isEmpty) field.name else s"$parentPath.${field.name}"

    field.dataType match {
      // 处理Struct类型：递归遍历内部所有字段
      case structType: StructType =>
        Array(struct(renameNestedColumns(structType, originalPath): _*).alias(newFieldName))
      // 处理Array类型：若元素是Struct则递归处理，否则直接重命名数组列
      case arrayType: ArrayType =>
        arrayType.elementType match {
          case elemStruct: StructType =>
            Array(transform(col(originalPath), 
              elem => struct(renameNestedColumns(elemStruct, "elem"): _*)
            ).alias(newFieldName))
          case _ =>
            Array(col(originalPath).alias(newFieldName))
        }
      // 普通数据类型直接重命名
      case _ =>
        Array(col(originalPath).alias(newFieldName))
    }
  }
}

使用方式 #

// 生成处理后的DataFrame
val correctedDF = scJsonDF.select(renameNestedColumns(scJsonDF.schema): _*)

// 验证处理后的Schema
correctedDF.printSchema()

代码逻辑说明 #

1. 递归遍历Schema的每个字段，先替换当前字段名的“-”为“_”
2. 遇到Struct类型时，递归处理其内部所有子字段，再重新构造Struct对象并重命名
3. 遇到Array类型时，若数组元素是Struct，用transform函数递归处理每个元素的Struct字段；若为普通类型则直接重命名数组列
4. 普通数据类型（String、Long等）直接通过列路径定位并完成重命名

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