如何将嵌套结构转换为 Spark DataFrame 的嵌套映射答案

【问题标题】：How to covert nested struct into nested map for Spark DataFrame如何将嵌套结构转换为 Spark DataFrame 的嵌套映射
【发布时间】：2021-11-04 14:50:00
【问题描述】：

我正在尝试批量写入 AWS DynamoDB，我必须在加载之前重新格式化数据帧，现在我的问题是如何将深度 structType 数据帧转换为 DynamoDB 可以识别的深度 Map 格式，并且不需要定义字段手动字段？

环境：Databricks 中的 Apache Spark 2.4.3/Spark 2.4.3、Scala 2.11、DynamoDB

来源具有如下深层结构

root
 |-- PK: string (nullable = false)
 |-- SK: string (nullable = false)
 |-- ee: struct (nullable = false)
 |    |-- kv: struct (nullable = false)
 |    |    |-- ss: map (nullable = true)
 |    |    |-- pp: struct (nullable = true)
 |    |    |    |-- gg: string (nullable = true)
 |    |    |    |-- nn: struct (nullable = true)
 |    |    |    |    |-- mm: string (nullable = true)
 |    |    |-- ll: array (nullable = true)
 |    |    |    |-- le: struct (containsNull = true)
 |    |    |    |    |-- lep: struct (nullable = true)

我找到了一些示例，但通常它们只能处理 1-2 级嵌套结构，但我的 dataFrame 在这种情况下“更深”。

【问题讨论】：

标签： scala apache-spark amazon-dynamodb

【解决方案1】：

以下函数将处理任何级别的深度嵌套数据帧。

val spark = SparkSession.builder().master("local[*]").getOrCreate()
  import org.apache.spark.sql.functions._
  spark.sparkContext.setLogLevel("ERROR")

// TODO: Instead of while/for loop, we can use pattern matching also.
def getFlattenDF(dataFrame: DataFrame): DataFrame = {
    var df = dataFrame
    var flag = true
    while (flag) {
      for ((name, types) <- df.dtypes) {
        if (types.startsWith("Array"))
          df = df.withColumn(name, explode_outer(col(name)))
        else if(types.startsWith("Map"))
          df = df.selectExpr("*", s"explode_outer($name)").drop(name)
        else if (types.startsWith("Struct"))
          df = df.selectExpr(Array("*") ++ df.select(s"$name.*").columns.map(s => s"$name" + "." + s + s" as $name" + s"_$s"): _*).drop(name)
      }
      flag = false
      for ((name, types) <- df.dtypes) {
        if (types.startsWith("Array") || types.startsWith("Struct") || types.startsWith("Map"))
          flag = true
      }
    }
    df
  }

 val df = //input dataframe

 getFlattenDF(df)

【讨论】：

尝试后出现错误“org.apache.spark.sql.AnalysisException: The number of aliases provided in the AS clause does not match the number of columns output by the UDTF expected 2 aliases但是得到了 "
是否可以在此处粘贴单行数据框？
这是被屏蔽的行：[[,,,,, [s,,,,,], [ww,,,,,], [ac -> [[, a8, fe]] , cd -> [[, 12, e]], vv -> [[, f6, f]], Cd -> [[, 12, true]]], [ss,, ee,,, 12,, [ ,, 12.12, 12.12],,, ee, 2t,,,,,, ss,,,,,,,], [cc,,,,,],,,,,, [ff,, 12,, 12 -12, [ss,, ff,,, dd], 12-12,],,,, [ss,, ss,,,, 1212,, [,, 12.1212, 12.12],,, uu, tt,,, ,,, ww,,,,,,,]]]]
和之前的错误追溯到代码“df = df.withColumn(name, explode_outer(col(name)))”
对于地图类型的列，如果我们在 withColumn 中使用它，则爆炸不起作用。但是如果在 select 子句中使用它，则为 map 类型分解会起作用。更新帖子，检查是否有效。

【解决方案2】：

这是我的最终解决方案：

/** convert from json to map, then wrap with AttributeValue class */
val tempMap = new ObjectMapper().readValue(testStringText, new TypeReference[JMap[String, Object]](){})
val testMap = toAttribute(tempMap)
new AttributeValue().withM(testMap.getM())

/** nest conversion - scala to java class, wrap with AttributeValue */
def toAttribute(m: Any): AttributeValue = {
    m match {
      case sm: java.util.LinkedHashMap[_, _] => {
        new AttributeValue().withM(sm.map(kv => (kv._1.toString, toAttribute(kv._2))).asJava)
      }
      case sl: java.util.ArrayList[_] => {
        new AttributeValue().withL(sl.map(item => toAttribute(item)).asJava.asInstanceOf[JCollection[AttributeValue]])
      }
      case st: String => new AttributeValue().withS(st)
      case bol: Boolean => new AttributeValue().withBOOL(bol)
      case dbl: java.lang.Double => new AttributeValue().withN(dbl.toString)
      case int: java.lang.Integer => new AttributeValue().withS(int.toString)
      case _ => {
        new AttributeValue()
      }
    }
}

【讨论】：