如何将一组字段写入 JSON？

Posted 2023-04-13

【中文标题】如何将一组字段写入 JSON？

【英文标题】：How to write a set of fields to JSON?

【发布时间】：2016-04-11 18:35:44

【问题描述】：

我正在尝试将我的数据框中的几个字段写入 JSON。我在数据框中的数据结构是

Key|col1|col2|col3|col4
key|a   |b   |c   |d
Key|a1  |b1  |c1  |d1

现在我正在尝试将 col1 到 col4 字段转换为 JSON 并为 Json 字段命名

预期输出

[Key,cols:[col1:a,col2:b,col3:c,col4:d,col1:a1,col2:b1,col3:c1,col4:d1]

我为此写了一个udf。

val summary = udf( 
(col1:String, col2:String, col3:String, col4:String) => "\"cols\":[" + "  \"col1\":" + col1 + ",\"col2\":" + col2 + ",\"col3\":" + col3 + ",\"col4\":" + col4 + "]"
)

val result = input.withColumn("Summary",summary('col1,'col2,'col3,'col4))
val result1 = result.select('Key,'Summary)
result1.show(10)

这是我的结果

[Key,cols:[col1:a,col2:b,col3:c,col4:d]]
[Key,cols:[col1:a1,col2:b1,col3:c1,col4:d1]]

如您所见，它们没有分组。有没有办法使用 UDF 本身对这些行进行分组。我是 scala/Spark 的新手，无法找出正确的 udf。

【问题讨论】：

我认为您没有正确终止您的“预期输出”；我希望最后会有另一个“]”来匹配开头的“[”。

【参考方案1】：

// Create your dataset
scala> val ds = Seq((1, "hello", 1L), (2, "world", 2L)).toDF("id", "token", "long")
ds: org.apache.spark.sql.DataFrame = [id: int, token: string ... 1 more field]

// select the fields you want to map to json
scala> ds.select('token, 'long).write.json("your-json")

// check the result
➜  spark git:(master) ✗ ls -ltr your-json/
total 16
-rw-r--r--  1 jacek  staff  27 11 kwi 17:18 part-r-00007-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 part-r-00006-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 part-r-00005-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 part-r-00004-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff  27 11 kwi 17:18 part-r-00003-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 part-r-00002-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 part-r-00001-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 part-r-00000-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
-rw-r--r--  1 jacek  staff   0 11 kwi 17:18 _SUCCESS
➜  spark git:(master) ✗ cat your-json/part-r-00003-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
"token":"hello","long":1
➜  spark git:(master) ✗ cat your-json/part-r-00007-91f81f62-54bb-42ae-bddc-33829a0e3c16.json
"token":"world","long":2

【参考方案2】：

UDF 会将一行映射到一行。如果您的DataFrame 中有多行要合并为一个元素，则需要使用像reduceByKey 这样的函数来聚合多行。

可能有一个 DataFrame 特定的函数来执行此操作，但我会使用 RDD 功能进行此处理，如下所示：

val colSummary = udf( 
(col1:String, col2:String, col3:String, col4:String) => "\"col1\":" + col1 + ",\"col2\":" + col2 + ",\"col3\":" + col3 + ",\"col4\":" + col4 + ""
)
val colRDD = input.withColumn("Summary",summary('col1,'col2,'col3,'col4)).rdd.map(x => (x.getString(0),x.getString(5)))

这给了我们一个RDD[(String,String)]，这将允许我们像reduceByKey一样使用PairRDDFunctions（参见docs）。元组的键是原始键，值是单个元素的 json 编码，我们需要将其聚合在一起以形成cols 列表。我们将它们全部粘合到一个逗号分隔的列表中，然后我们添加开头和结尾，然后我们就完成了。

val result = colRDD.reduceByKey((x,y) => (x+","+y)).map(x => "["+x._1+",\"cols\":["+x._2+"]]")
result.take(10)

【讨论】：

谢谢。它成功了。只需在答案中进行一些编辑（在此评论中解释）。 val colRDD 最后缺少一个 ')'。并且结果最后缺少一个“）”。在 colRDD 中，它是 x.getString(5) 而不是 x.getString(1)，因为在我们向输入数据帧添加字段后，摘要将是第 5 个字段。最后 result.take(10) as 'result' 是一个 rdd。