我有一个具有以下模式的Dataframe。

root
 |-- id: string (nullable = true)
 |-- type: string (nullable = true)
 |-- key: string (nullable = true)
 |-- value: string (nullable = true)

它首先以“value”列为轴心。

val pivot_df = df.groupBy("id","key").pivot("type").agg(first("value"))

“value”列的数据可能包含一组嵌套的json。所以之前我正在转换整个“value”列以获得所需的模式。

val schema = spark.read.json(df.select("value").as[String]).schema

val res_df = df.select($"id",$"type",$"key",from_json($"value",schema).as("s")).select("id","type","key","s.*")

但现在我需要首先将它放在“value”列数据上，然后从json读取所有新生成的列。有人能建议应该应用哪些代码更改吗？