将数据帧分组到列表中

Posted 2023-04-17

【中文标题】将数据帧分组到列表中

【英文标题】：Spark dataframes groupby into list

【发布时间】：2015-08-06 20:00:58

【问题描述】：

我正在尝试对集合进行一些分析。我有一个示例数据集，如下所示：

orders.json

"items":[1,2,3,4,5]
"items":[1,2,5]
"items":[1,3,5]
"items":[3,4,5]

它只是一个字段，它是代表 ID 的数字列表。

这是我尝试运行的 Spark 脚本：

val sparkConf = new SparkConf()
  .setMaster("local[*]")
  .setAppName("Dataframe Test")

val sc = new SparkContext(sparkConf)
val sql = new SQLContext(sc)

val dataframe = sql.read.json("orders.json")

val expanded = dataframe
  .explode[::[Long], Long]("items", "item1")(row => row)
  .explode[::[Long], Long]("items", "item2")(row => row)

val grouped = expanded
  .where(expanded("item1") !== expanded("item2"))
  .groupBy("item1", "item2")
  .count()

val recs = grouped
  .groupBy("item1")

创建expanded 和grouped 很好，简而言之expanded 是两个ID 的所有可能集合的列表，其中两个ID 在同一个原始集合中。 grouped 过滤掉与自己匹配的 ID，然后将所有唯一的 ID 对组合在一起，并为每个 ID 生成一个计数。 grouped 的架构和数据样本为：

root
 |-- item1: long (nullable = true)
 |-- item2: long (nullable = true)
 |-- count: long (nullable = false)

[1,2,2]
[1,3,2]
[1,4,1]
[1,5,3]
[2,1,2]
[2,3,1]
[2,4,1]
[2,5,2]
...

所以，我的问题是：我现在如何对每个结果中的第一项进行分组，以便获得一个元组列表？对于上面的示例数据，我希望类似于以下内容：

[1, [(2, 2), (3, 2), (4, 1), (5, 3)]]
[2, [(1, 2), (3, 1), (4, 1), (5, 2)]]

正如您在我的脚本中看到的 recs 所见，我认为您应该首先对“item1”进行 groupBy，这是每行中的第一项。但在那之后，您将得到这个 GroupedData 对象，该对象的操作非常有限。真的，您只需要进行 sum、avg 等聚合。我只想列出每个结果中的元组。

此时我可以轻松使用 RDD 函数，但这与使用 Dataframes 不同。有没有办法用数据框函数做到这一点。

【参考方案1】：

您可以使用自 1.6 起可用的 org.apache.spark.sql.functions（collect_list 和 struct）构建它

val recs =grouped.groupBy('item1).agg(collect_list(struct('item2,'count)).as("set"))


+-----+----------------------------+
|item1|set                         |
+-----+----------------------------+
|1    |[[5,3], [4,1], [3,2], [2,2]]|
|2    |[[4,1], [1,2], [5,2], [3,1]]|
+-----+----------------------------+

你也可以使用collect_set

编辑：有关信息，tuples 不存在于数据框中。最接近的结构是 struct，因为它们相当于无类型数据集 API 中的案例类。

编辑 2：还要注意collect_set 附带警告，即结果实际上不是集合（SQL 类型中没有具有集合属性的数据类型）。这意味着您最终可能会得到不同的“集合”，它们的顺序不同（至少在 2.1.0 版中）。然后有必要使用sort_array 对它们进行排序。

【讨论】：

collect_list() 包含所有元素，而 collect_set() 包含不同的元素