将 spark 数据帧的所有列转换为 json 格式，然后将 json 格式的数据作为列包含在另一个/父数据帧中

Question

使用 df.toJSON 将数据帧（比如子数据帧）转换为 json

json 转换后的架构如下所示：

root
 |-- value: string (nullable = true)

我使用以下建议将子数据框放入中间父架构/数据框：

scala> parentDF.toJSON.select(struct($"value").as("data")).printSchema
root
 |-- data: struct (nullable = false)
 |    |-- value: string (nullable = true)

现在我仍然需要进一步构建 parentDF 架构以使其看起来像：

root
 |-- id
 |-- version 
 |-- data: struct (nullable = false)
 |    |-- value: string (nullable = true)

Q1) 如何使用 value 中的 id 构建 id 列（即 value.id 需要表示为 id）

Q2）我需要从不同的数据框（比如 versionDF）中获取版本，其中版本是一个常量（在所有列中）。我是否从此 versionDF 中获取一行以读取 version 列的值，然后将其填充为 parentDF 中的文字？

请帮助解决这方面的任何代码 sn-ps。

Answer 1

在select 语句中使用to_json 而不是toJSON 并选择所需的列以及to_json 函数。

检查下面的代码。

val version =  // Get version value from versionDF
parentDF.select($"id",struct(to_json(struct($"*")).as("value")).as("data"),lit(version).as("version"))

scala> parentDF.select($"id",struct(to_json(struct($"*")).as("value")).as("data"),lit(version).as("version")).printSchema
root
 |-- id: integer (nullable = false)
 |-- data: struct (nullable = false)
 |    |-- value: string (nullable = true)
 |-- version: double (nullable = false)

更新

scala> df.select($"id",to_json(struct(struct($"*").as("value"))).as("data"),lit(version).as("version")).printSchema
root
 |-- id: integer (nullable = false)
 |-- data: string (nullable = true)
 |-- version: integer (nullable = false)


scala> df.select($"id",to_json(struct(struct($"*").as("value"))).as("data"),lit(version).as("version")).show(false)
+---+------------------------------------------+-------+
|id |data                                      |version|
+---+------------------------------------------+-------+
|1  |{"value":{"id":1,"col1":"a1","col2":"b1"}}|1      |
|2  |{"value":{"id":2,"col1":"a2","col2":"b2"}}|1      |
|3  |{"value":{"id":3,"col1":"a3","col2":"b3"}}|1      |
+---+------------------------------------------+-------+

Update-1

scala> df.select($"id",to_json(struct($"*").as("value")).as("data"),lit(version).as("version")).printSchema
root
 |-- id: integer (nullable = false)
 |-- data: string (nullable = true)
 |-- version: integer (nullable = false)


scala> df.select($"id",to_json(struct($"*").as("value")).as("data"),lit(version).as("version")).show(false)
+---+--------------------------------+-------+
|id |data                            |version|
+---+--------------------------------+-------+
|1  |{"id":1,"col1":"a1","col2":"b1"}|1      |
|2  |{"id":2,"col1":"a2","col2":"b2"}|1      |
|3  |{"id":3,"col1":"a3","col2":"b3"}|1      |
+---+--------------------------------+-------+

Answer 2

试试这个：

scala> val versionDF = List((1.0)).toDF("version")
versionDF: org.apache.spark.sql.DataFrame = [version: double]

scala> versionDF.show
+-------+
|version|
+-------+
|    1.0|
+-------+


scala> val version = versionDF.first.get(0)
version: Any = 1.0

scala>

scala> val childDF = List((1,"a1","b1"),(2,"a2","b2"),(3,"a3","b3")).toDF("id","col1","col2")
childDF: org.apache.spark.sql.DataFrame = [id: int, col1: string ... 1 more field]

scala> childDF.show
+---+----+----+
| id|col1|col2|
+---+----+----+
|  1|  a1|  b1|
|  2|  a2|  b2|
|  3|  a3|  b3|
+---+----+----+


scala>

scala> val parentDF =  childDF.toJSON.select(struct($"value").as("data")).withColumn("id",from_json($"data.value",childDF.schema).getItem("id")).withColumn("version",lit(version))
parentDF: org.apache.spark.sql.DataFrame = [data: struct<value: string>, id: int ... 1 more field]

scala> parentDF.printSchema
root
 |-- data: struct (nullable = false)
 |    |-- value: string (nullable = true)
 |-- id: integer (nullable = true)
 |-- version: double (nullable = false)

scala> parentDF.show(false)
+----------------------------------+---+-------+
|data                              |id |version|
+----------------------------------+---+-------+
|[{"id":1,"col1":"a1","col2":"b1"}]|1  |1.0    |
|[{"id":2,"col1":"a2","col2":"b2"}]|2  |1.0    |
|[{"id":3,"col1":"a3","col2":"b3"}]|3  |1.0    |
+----------------------------------+---+-------+