我正在尝试将数据框中的几个字段写入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。
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)]，这将允许我们使用PairRDDFunctionslike reduceByKey（请参阅docs）。元组的键是原始键，值是我们需要聚合在一起以组成cols列表的单个元素的json编码。我们将它们全部粘贴到一个逗号分隔的列表中，然后添加开始和结束，然后完成。
val result = colRDD.reduceByKey((x,y) => (x+","+y)).map(x => "["+x._1+",{\"cols\":["+x._2+"]}]")
result.take(10)
本文收集自互联网，转载请注明来源。
如有侵权，请联系[email protected] 删除。
我来说两句