如何将聚合数据添加到 Apache Spark 中的原始数据集中？

我试图弄清楚如何聚合数据集中的数据，然后使用 Apache Spark 将结果添加到原始数据集。 我已经尝试了 2 个我不满意的解决方案，我想知道是否有一个我没有看到的更具可扩展性和更高效的解决方案。

以下是输入和预期输出数据的非常简化的示例：

Input:

客户列表，以及每个客户的购买商品列表。

(John, [toast, butter])
(Jane, [toast, jelly])

Output:

客户列表、每个客户的已购买商品列表以及每个商品的购买该商品的客户数量。

(John, [(toast, 2), (butter, 1)])
(Jane, [(toast, 2), (jelly, 1)])

以下是我迄今为止尝试过的解决方案，列出了步骤和输出数据。

解决方案#1：

Start with a pair rdd:
(John, [toast, butter])
(Jane, [toast, jelly])

flatMapToPair:
(toast, John)
(butter, John)
(toast, Jane)
(jelly, Jane)

aggregateByKey: 
(toast, [John, Jane])
(butter, [John])
(jelly, [Jane])

flatMapToPair: (using the size of the list of customers)
(John, [(toast, 2), (butter, 1)])
(Jane, [(toast, 2), (jelly, 1)])

虽然这适用于小型数据集，但对于较大的数据集来说这是一个糟糕的想法，因为在某一时刻，您为每个产品保存了一个巨大的客户列表，这些客户列表可能无法放入执行器的内存中。

解决方案#2：

Start with a pair rdd:
(John, [toast, butter])
(Jane, [toast, jelly])

flatMapToPair:
(toast, John)
(butter, John)
(toast, Jane)
(jelly, Jane)

aggregateByKey: (counting customers without creating a list)
(toast, 2)
(butter, 1)
(jelly, 1)

join: (using the two previous results)
(toast, (John, 2))
(butter, (John, 1))
(toast, (Jane, 2))
(jelly, (Jane, 1))

mapToPair:
(John, (toast, 2))
(John, (butter, 1))
(Jane, (toast, 2))
(Jane, (jelly, 1))

aggregateByKey:
(John, [(toast, 2), (butter, 1)])
(Jane, [(toast, 2), (jelly, 1)])

这个解决方案应该可行，但我觉得应该有一些其他可能不涉及加入 RDD 的解决方案。

对于这个问题是否有更可扩展/更高效/更好的“解决方案#3”？

这是一个dataframe供你尝试和玩耍的方式

如果您已经有配对rdds然后打电话toDF与列名应该给你一个dataframe as

val df = pairedRDD.toDF("key", "value")

应该是

+----+---------------+
|key |value          |
+----+---------------+
|John|[toast, butter]|
|Jane|[toast, jelly] |
+----+---------------+

现在你所要做的就是explode, groupby, 计数的聚合然后再次explode, groupby and 聚合以获得带有计数的原始数据集 as

import org.apache.spark.sql.functions._
df.withColumn("value", explode(col("value")))
  .groupBy("value").agg(count("value").as("count"), collect_list("key").as("key"))
  .withColumn("key", explode(col("key")))
  .groupBy("key").agg(collect_list(struct("value", "count")).as("value"))

这应该给你

+----+-----------------------+
|key |value                  |
+----+-----------------------+
|John|[[toast,2], [butter,1]]|
|Jane|[[jelly,1], [toast,2]] |
+----+-----------------------+

您可以进一步处理dataframe或改回rdd通过使用.rdd api.