Scala: преобразование столбца dict для данных в таблицу

Question

У меня есть датафрейм rating в формате ниже:

 id         | percentile
------------+-------------------------------------------------
 0011111111 | {"80": 3438, "40": 1063, "60": 2119, "20": 620}

Мне нужно преобразовать в формат ниже:

 id         | 80   | 40   | 60   | 20   |
------------+------+------+------+------+
 0011111111 | 3438 | 1063 | 2119 | 620  |

Я пробовал с кодом ниже, но не помогло:

val schema = StructType(Seq(
       StructField("80", DoubleType, true)
      ,StructField("60", DoubleType, true)
      ,StructField("40", DoubleType, true)
      ,StructField("20", DoubleType, true)
    ))

val rating1 = rating.withColumn("jsonData", from_json(col("percentile"), schema))
rating1.show()

+--------------------+--------------------+--------------------+
|                 cid|          percentile|            jsonData|
+--------------------+--------------------+--------------------+
|          0011111111|{"80": 3438, "40"...|[3438.0, 1063.0, ...|

Как мне получить 80, 60, 40, 20 в виде столбцов

Answer 1

вы можете достичь этого, используя функцию from_json, я думаю, что это можно сделать без выбора каждого элемента 'map.getField ("80"), "map.getField (" 40 "), ... и передачиих как массив [столбец]

  val str = new StructType()
    .add("80","string")
    .add("40","string")
    .add("60","string")
    .add("20","string")

  df.select('id, from_json('percentile,str).as("map"))
    .select('id,
      'map.getField("80"),
      'map.getField("40"),
      'map.getField("60"),
      'map.getField("20")
    ).show()

    +----------+-----------------------------------------------+
    |id        |percentile                                     |
    +----------+-----------------------------------------------+
    |0011111111|{"80": 3438, "40": 1063, "60": 2119, "20": 620}|
    +----------+-----------------------------------------------+

    root
     |-- id: string (nullable = true)
     |-- percentile: string (nullable = true)

    +----------+------+------+------+------+
    |        id|map.80|map.40|map.60|map.20|
    +----------+------+------+------+------+
    |0011111111|  3438|  1063|  2119|   620|
    +----------+------+------+------+------+

Answer 2

scala> df.show(false)
+--------+-----------------------------------------------+
|id      |percentile                                     |
+--------+-----------------------------------------------+
|11111111|{"80": 3438, "40": 1063, "60": 2119, "20": 620}|
+--------+-----------------------------------------------+

//UDF to replace '{' and '}' from column percentile

scala> import org.apache.spark.sql.expressions.{UserDefinedFunction} 
scala> val replace = udf((data: String , rep : String)=>data.replaceAll(rep, ""))
scala> val df1 = df.withColumn("percentile", replace($"percentile", lit("\\{"))).withColumn("percentile", replace($"percentile", lit("\\}")))

scala> df1.show(false)
+--------+---------------------------------------------+
|id      |percentile                                   |
+--------+---------------------------------------------+
|11111111|"80": 3438, "40": 1063, "60": 2119, "20": 620|
+--------+---------------------------------------------+

//Mapping header with its value from column percentile

scala> val df2 = df1.withColumn("var", explode(split(col("percentile"), ", "))).withColumn("header", split(col("var"), ": ")(0)).withColumn("value", split(col("var"), ": ")(1)).drop("var")

scala> df2.show(false)
+--------+---------------------------------------------+------+-----+
|id      |percentile                                   |header|value|
+--------+---------------------------------------------+------+-----+
|11111111|"80": 3438, "40": 1063, "60": 2119, "20": 620|"80"  |3438 |
|11111111|"80": 3438, "40": 1063, "60": 2119, "20": 620|"40"  |1063 |
|11111111|"80": 3438, "40": 1063, "60": 2119, "20": 620|"60"  |2119 |
|11111111|"80": 3438, "40": 1063, "60": 2119, "20": 620|"20"  |620  |
+--------+---------------------------------------------+------+-----+

//Final expected output
scala> df2.groupBy("id").pivot("header").agg(concat_ws("",collect_list(col("value")))).show()
+--------+----+----+----+----+
|      id|"20"|"40"|"60"|"80"|
+--------+----+----+----+----+
|11111111| 620|1063|2119|3438|
+--------+----+----+----+----+