Spark scala - 嵌套StructType轉換爲地圖

我在scala中使用Spark 1.6。Spark scala - 嵌套StructType轉換爲地圖

我使用對象在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指數。

星火下面的模式讀取索引：

|-- 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的地圖，但是我有以下錯誤：

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代碼片段：

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到一個地圖？我已經閱讀過看到文檔中可用的toMap方法，但無法找到如何使用它（不熟悉隱式參數），因爲我是一名scala初學者。

由於提前，

來源

2017-10-04 jam_ab

可以請你加UDF代碼片段？ –

UDF不會幫助很多，因爲我只是試圖獲得一個Map所需的Map [String，Map [String，String]]。 –

val getSubField：Map [String，Map [String，String]] => String =（params：Map [String，Map [String，String]]）=> {0128} ）getOrElse（「p1_detail」，null.asInstanceOf [字符串]） \t RETURN_STRING }' –

不能指定PARAM作爲StructType對象的類型，而不是指定類型行。

//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; 
    }

我希望這有助於！

來源

2017-10-05 16:05:32

我會盡力而爲，非常感謝您的幫助。 –

我終於解開了它，如下所示：

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 map_temp = convertRowToMap[Row](row) 

    val map_to_return = map_temp.map{case(k,v) => k -> convertRowToMap[String](v)} 

    map_to_return 
} 
    val udfrowToMap = udf(rowToMap)

來源

2017-10-09 15:27:31

Spark scala - 嵌套StructType轉換爲地圖

回答

相關問題