scala - 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 坐标，即 double 对。我想为每个 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

You currently can't use SparkContext inside a Spark task, so in this case you'd have to call some kind of local K-means library. One example you can try to use is Weka (http://www.cs.waikato.ac.nz/ml/weka/). You can then load your text files as an RDD of strings with SparkContext.wholeTextFiles and call Weka on each one.