如何在 Spark 中的 groupby 之后运行 RDD 操作？

Question

我有大量数据要对其执行聚类。问题是，我不希望对整个集合进行一个聚类，而是对每个用户进行一个聚类。基本上我会先做一个 groupby 用户 ID，然后运行 ​​KMeans。

问题是，一旦您执行 groupby，任何映射都会在 spark 控制器上下文之外，因此任何创建 RDD 的尝试都会失败。在 mllib 中 Spark 的 KMeans 库需要一个 RDD（因此它可以并行化）。

我看到了两种解决方法，但我希望有更好的解决方案。

1) 手动遍历控制器中的所有数千个用户（当事情变大时可能是数百万个），并为每个用户运行 kmeans。

2) 在控制器中进行 groupby，然后在 map 中运行由外部库提供的非并行 kmeans。

请告诉我还有另一种方法，我宁愿拥有一切 ||尽可能。

Answer 1

编辑：在回复时我不知道是pyspark。但是，我将把它留作一个可以改编的想法

我遇到了类似的问题，并且能够提高性能，但这对我来说仍然不是理想的解决方案。也许对你来说它可以工作。

这个想法是将 RDD 分解为许多较小的 RDD（每个用户 id 一个新的 RDD），将它们保存到一个数组中，然后为每个“子 RDD”调用处理函数（在您的情况下为集群）。建议的代码如下（cmets 中的解释）：

// A case class just to use as example
case class MyClass(userId: Long, value: Long, ...)

// A Scala local array with the user IDs (Could be another iterator, such as List or Array):
val userList: Seq[Long] = rdd.map{ _.userId }.distinct.collect.toSeq  // Just a suggestion!

// Now we can create the new rdds:
val rddsList: Seq[RDD[MyClass]] = userList.map { 
  userId => rdd.filter({ item: MyClass => item.userId == userId }) 
}.toSeq

// Finally, we call the function we want for each RDD, saving the results in a new list. 
// Note the ".par" call, which is used to start the expensive execution for multiple RDDs at the same time
val results = rddsList.par.map {
  r => myFunction(r)
}

我知道这与您的第一个选项大致相同，但是通过使用 .par 调用，我能够提高性能。

此调用将rddsList 对象转换为ParSeq 对象。这个新的 Scala 对象允许并行计算，因此，理想情况下，map 函数将同时为多个 RDD 调用 myFunction(r)，这样可以提高性能。

更多并行集合详情请查看Scala Documentation。