这是一个示例数据
val df4 = sc.parallelize(List(
("A1",45, "5", 1, 90),
("A2",60, "1", 1, 120),
("A6", 30, "9", 1, 450),
("A7", 89, "7", 1, 333),
("A7", 89, "4", 1, 320),
("A2",60, "5", 1, 22),
("A1",45, "22", 1, 1)
)).toDF("CID","age", "children", "marketplace_id","value")
感谢@shu提供了这段代码
val df5 = df4.selectExpr("CID","""to_json(named_struct("id", children)) as item""", "value", "marketplace_id")
+---+-----------+-----+--------------+
|CID|item |value|marketplace_id|
+---+-----------+-----+--------------+
|A1 |{"id":"5"} |90 |1 |
|A2 |{"id":"1"} |120 |1 |
|A6 |{"id":"9"} |450 |1 |
|A7 |{"id":"7"} |333 |1 |
|A7 |{"id":"4"} |320 |1 |
|A2 |{"id":"5"} |22 |1 |
|A1 |{"id":"22"}|1 |1 |
+---+-----------+-----+--------------+
当你这么做的时候 df5.dtypes
```
(CID,StringType), (item,StringType), (value,IntegerType), (marketplace_id,IntegerType)
列项是字符串类型的,有没有一种方法可以是json/对象类型的(如果是这样的话)?
编辑1:我将在这里描述我试图实现的目标,以上两个步骤保持不变。
val w = Window.partitionBy("CID").orderBy(desc("value"))
val sorted_list = df5.withColumn("item", collect_list("item").over(w)).groupBy("CID").agg(max("item") as "item")
输出:
+---+-------------------------+
|CID|item |
+---+-------------------------+
|A6 |[{"id":"9"}] |
|A2 |[{"id":"1"}, {"id":"5"}] |
|A7 |[{"id":"7"}, {"id":"4"}] |
|A1 |[{"id":"5"}, {"id":"22"}]|
+---+-------------------------+
不管里面是什么 `[ ]` 是字符串。这给我们正在使用的一个工具带来了问题。
对不起,对不起,我是新来的斯卡拉,如果这是一个基本的问题。
2条答案
按热度按时间z9smfwbn1#
商店
json
数据使用struct
键入,检查下面的代码。7gcisfzg2#
根据您对将数据集转换为json所做的注解,您将使用:
输出如下所示:
如果需要将内存中的内容传递给函数,而不是
write().json(...)
使用toJSON