Apache Spark K-Means 聚类 - 用于输入的 RDD

Question

我正在尝试对分组数据运行 Spark 的 k-means 聚类，但是当我尝试对每个组进行聚类时遇到了各种错误。

输入的 RDD 看起来像 (userID: Long, coords: [Vector]) 即：

org.apache.spark.rdd.RDD[(Long, Seq[org.apache.spark.mllib.linalg.Vector])]

向量包含 X Y 坐标，即双精度对。我想为每个 UserID 识别坐标集群，所以我在 RDD 上进行映射，并尝试为每个组运行 k-means：

val userClusters = userCoordVectors.map {
  case (userId, coords) =>
    val clusters = 4
    val iterations = 30
    // Need to convert coords to RDD for input into K-Means
    val parsedData = sc.parallelize(coords)

    // Apply k-means
    val model = KMeans.train(parsedData, clusters, iterations)
    ...
    etc
}

但是当我运行它时，我从该行得到一个 NPE：

val parsedData = sc.parallelize(coords)

问题是，我必须将坐标转换为 RDD 以进行 K-Means 操作。

另一方面，如果我先收集输入 RDD，那么我就得不到 NPE。相反，我得到一个 Java 堆错误，大概是因为我正在实现整个 RDD。

 val userClusters = sc.parallelize(userCoordVectors.collect.map {
 ...
 })

在这个 RDD 中收集数据在这里似乎是错误的，所以我假设应该有更好的方法，但我不知道如何让 parsedData 行工作。

谁能看到我在这里尝试使用 K-Means 的方式有任何明显的错误，或者建议如何实现在每个组中对我的数据进行聚类的目标？

Answer 1

您不能在 RDD 运算符的任何函数中使用 SparkContext 或 RDD。它们不能被序列化并通过网络发送。

Matei Zaharia 在这里回答：http://apache-spark-user-list.1001560.n3.nabble.com/Can-we-get-a-spark-context-inside-a-mapper-td9605.html

您目前无法在 Spark 任务中使用 SparkContext，因此在这种情况下，您必须调用某种本地 K-means 库。您可以尝试使用的一个示例是 Weka (http://www.cs.waikato.ac.nz/ml/weka/)。然后，您可以使用 SparkContext.wholeTextFiles 将文本文件加载为字符串的 RDD，并在每个文件上调用 Weka。