13

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

订单.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 函数，但这与使用 Dataframe 不同。有没有办法用数据框函数做到这一点。

4

1 回答 1

16

您可以使用org.apache.spark.sql.functions(collect_list和struct) 从 1.6 开始构建它

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然后有必要对它们进行排序。

于 2016-10-18T09:39:47.677 回答