SPARK RDD - Clustering - K-Means

Question

Imagine that I've this dataset (just a sample)

A   B   C
1   23  1000
2   52  5000
3   12  500
4   10  450

I'm trying to assign each row to a clustering based on C value. Like this:

A   B   C   CLUSTER
1   23  1000    2
2   52  5000    1
3   12  500     3
4   10  450     3

For that I'm using K-Means algorithm using Spark:

import org.apache.spark.mllib.clustering.{KMeans, KMeansModel}
import org.apache.spark.mllib.linalg.Vectors

val data = sc.textFile("/user/cloudera/TESTE1")
val parsedData = data.map(s => Vectors.dense(s.split(',').map(_.toDouble))).cache()

val numClusters = 4
val numIterations = 20
val clusters = KMeans.train(parsedData, numClusters, numIterations)

val WSSSE = clusters.computeCost(parsedData)
println("Within Set Sum of Squared Errors = " + WSSSE)

clusters.save(sc, "/user/cloudera/KMeansModel")
val sameModel = KMeansModel.load(sc, "/user/cloudera/KMeansModel")

But this script extracts me a .gz.parquet file and when I try to see what type of information this file contains, using:

sqlContext.read.parquet("/user/cloudera/KMeansModel/data/part-r-00000-f551ea29-54db-45be-8cba-d06a97d6d9f2.gz.parquet").show

I'm getting this:

+---+--------------------+
| id|               point|
+---+--------------------+
|  0|[9.39601519885208...|
|  1|[9.80112351958380...|
|  2|[9.63822872186722...|
|  3|[9.44194658832542...|
+---+--------------------+

How can I get the table that I put above? Basically I just want to extract the same fields and add a column with the cluster calculated by K-Mwans to each row...

Many thanks!

Answer 1

This is well answered in this earlier question:

https://stackoverflow.com/q/31447141/1060350

Beware that Spark k-means is slow.

If your data fits into main memory (i.e. a few gigabyte, which means billions of vectors!) then other tools such as ELKI that don't have the cluster overhead will be much faster. Use spark only for preprocessing the data, if you e.g. have several TB of jsons, and you need to first extract the numbers out of the JSONs, then here is where Spark shines.

Once your data is then vectors, use ELKI instead, it's much faster.