如何重命名 DataFrame 中与嵌套 JSON 对应的字段

我正在尝试使用以下方法处理移动应用程序中收到的 JSON 事件（如点击等）spark 1.5.2。有多个应用程序版本，并且事件的结构因版本而异。

假设版本 1 具有以下结构：

{
    "timestamp": "",
    "ev": {
                "app": {
                    "appName": "XYZ",
                    "appVersion": "1.2.0"
                }
                "device": {
                    "deviceId": "ABC",
                    ...
                }
                ...
             }
}

另一个版本的结构如下：

{
    "timestamp": "",
    "ev": {
                "_a": {
                    "name": "XYZ",
                    "version": "1.3.0"
                }
                "_d": {
                    "androidId": "ABC",
                    ...
                }
                ...
             }
}

我希望能够为结构创建单个数据框并执行一些查询。

我正在使用以下方法为每个结构创建两个不同的数据框filter功能。现在我需要能够重命名列以对两个数据帧执行联合操作。

我在用：

df.withColumnRenamed("ev.app", "ev._a").withColumnRenamed("ev.device", "ev._d");

但这是行不通的。我该如何实现这一目标？

如果只是重命名嵌套列而不是更改模式结构，那么替换 DataFrame 模式（使用新模式重新创建 DataFrame）就可以了。

object functions {

  private def processField(structField: StructField, fullColName: String, oldColName: String, newColName: String): StructField = {
    if (fullColName.equals(oldColName)) {
      new StructField(newColName, structField.dataType, structField.nullable)
    } else if (oldColName.startsWith(fullColName)) {
      new StructField(structField.name, processType(structField.dataType, fullColName, oldColName, newColName), structField.nullable)
    } else {
      structField
    }
  }

  private def processType(dataType: DataType, fullColName: String, oldColName: String, newColName: String): DataType = {
    dataType match {
      case structType: StructType =>
        new StructType(structType.fields.map(
          f => processField(f, if (fullColName == null) f.name else s"${fullColName}.${f.name}", oldColName, newColName)))
      case other => other
    }
  }

  implicit class ExtDataFrame(df: DataFrame) {
    def renameNestedColumn(oldColName: String, newColName: String): DataFrame = {
      df.sqlContext.createDataFrame(df.rdd, processType(df.schema, null, oldColName, newColName).asInstanceOf[StructType])
    }
  }
}

Usage:

scala> import functions._
import functions._

scala> df.printSchema
root
 |-- geo_info: struct (nullable = true)
 |    |-- city: string (nullable = true)
 |    |-- country_code: string (nullable = true)
 |    |-- state: string (nullable = true)
 |    |-- region: string (nullable = true)

scala> df.renameNestedColumn("geo_info.country_code", "country").printSchema
root
 |-- geo_info: struct (nullable = true)
 |    |-- city: string (nullable = true)
 |    |-- country: string (nullable = true)
 |    |-- state: string (nullable = true)
 |    |-- region: string (nullable = true)

这个实现是递归的，所以它也应该处理这样的情况：

df.renameNestedColumn("a.b.c.d.e.f", "bla")