Spark scala - 嵌套结构类型转换为地图答案

【问题标题】：Spark scala - Nested StructType conversion to MapSpark scala - 嵌套结构类型转换为地图
【发布时间】：2018-03-15 22:28:35
【问题描述】：

我在 scala 中使用 Spark 1.6。

我在 ElasticSearch 中用一个对象创建了一个索引。对象“params”被创建为 Map[String, Map[String, String]]。示例：

val params : Map[String, Map[String, String]] = ("p1" -> ("p1_detail" -> "table1"), "p2" -> (("p2_detail" -> "table2"), ("p2_filter" -> "filter2")), "p3" -> ("p3_detail" -> "table3"))

这给了我如下所示的记录：

{
        "_index": "x",
        "_type": "1",
        "_id": "xxxxxxxxxxxx",
        "_score": 1,
        "_timestamp": 1506537199650,
        "_source": {
           "a": "toto",
           "b": "tata",
           "c": "description",
           "params": {
              "p1": {
                 "p1_detail": "table1"
              },
              "p2": {
                 "p2_detail": "table2",
                 "p2_filter": "filter2"
              },
              "p3": {
                 "p3_detail": "table3"
              }
           }
        }
     },

然后我尝试读取 Elasticsearch 索引以更新值。

Spark 使用以下架构读取索引：

|-- a: string (nullable = true)
|-- b: string (nullable = true)
|-- c: string (nullable = true)
|-- params: struct (nullable = true)
|    |-- p1: struct (nullable = true)
|    |    |-- p1_detail: string (nullable = true)
|    |-- p2: struct (nullable = true)
|    |    |-- p2_detail: string (nullable = true)
|    |    |-- p2_filter: string (nullable = true)
|    |-- p3: struct (nullable = true)
|    |    |-- p3_detail: string (nullable = true)

我的问题是对象被读取为结构。为了管理和轻松更新字段，我想要一个 Map，因为我对 StructType 不是很熟悉。

我尝试将 UDF 中的对象作为 Map 获取，但出现以下错误：

 User class threw exception: org.apache.spark.sql.AnalysisException: cannot resolve 'UDF(params)' due to data type mismatch: argument 1 requires map<string,map<string,string>> type, however, 'params' is of struct<p1:struct<p1_detail:string>,p2:struct<p2_detail:string,p2_filter:string>,p3:struct<p3_detail:string>> type.;

UDF 代码 sn-p:

val getSubField : Map[String, Map[String, String]] => String = (params : Map[String, Map[String, String]]) => { val return_string = (params ("p1") getOrElse("p1_detail", null.asInstanceOf[String]) return_string }

我的问题：我们如何将这个 Struct 转换为 Map？我已经阅读了文档中可用的 toMap 方法，但由于我是 Scala 初学者，所以找不到如何使用它（对隐式参数不太熟悉）。

提前致谢，

【问题讨论】：

能否添加UDF代码sn-p？
UDF 不会有太大帮助，因为我只是想获得一个 Map[String, Map[String, String]] 应该是 Struct 的地方。
val getSubField : Map[String, Map[String, String]] => String = (params : Map[String, Map[String, String]]) => { val return_string = (params ("p1") getOrElse("p1_detail", null.asInstanceOf[String]) return_string }
我可以用 SructType 替换“params”类型，但我就是不知道如何将其转换为 Map。
您不能将 StructType 对象指定为参数，因为它们表示集合模式，而是将类型定义为 Row。检查我的答案。

标签： scala apache-spark elasticsearch spark-dataframe

【解决方案1】：

我终于解决了如下：

def convertRowToMap[T](row: Row): Map[String, T] = {
  row.schema.fieldNames
    .filter(field => !row.isNullAt(row.fieldIndex(field)))
    .map(field => field -> row.getAs[T](field))
    .toMap
}

/* udf that converts Row to Map */
val rowToMap: Row => Map[String, Map[String, String]] = (row: Row) => {
  val mapTemp = convertRowToMap[Row](row)
  
  val mapToReturn = mapTemp.map { case (k, v) => k -> convertRowToMap[String](v) }
  
  mapToReturn   
}
val udfrowToMap = udf(rowToMap)

【讨论】：

谢谢队友，你的解决方案帮我解决了我的问题。

【解决方案2】：

您不能将参数的类型指定为 StructType 对象，而是将类型指定为 Row。

//Schema of parameter
def schema:StructType = (new StructType).add("p1", (new StructType).add("p1_detail", StringType))
      .add("p2", (new StructType).add("p2_detail", StringType).add("p2_filter",StringType))
      .add("p3", (new StructType).add("p3_detail", StringType))

 //Not allowed
 val extractVal: schema => collection.Map[Nothing, Nothing] = _.getMap(0)

解决方案：

// UDF example to process struct column
val extractVal: (Row) => collection.Map[Nothing, Nothing] = _.getMap(0)

// You would implement something similar
   val getSubField : Map[String, Map[String, String]] => String =
  (params : Row) =>
  {
    val p1 = params.getAs[Row]("p1")
    .........
    return null;
  }

我希望这会有所帮助！

【讨论】：