Compute Cost of Kmeans
<p>I am using this <a href="https://github.com/yahoo/lopq/blob/master/python/lopq/model.py" rel="nofollow">model</a>, which is not written by me. In order to predict the centroids I had to do this:</p>

<pre><code>model = cPickle.load(open("/tmp/model_centroids_128d_pkl.lopq"))
codes = d.map(lambda x: (x[0], model.predict_coarse(x[1])))
</code></pre>

<p>where `d.first()' yields this:</p>

<pre><code>(u'3768915289',
 array([ -86.00641097, -100.41325623,   &lt;128 coords in total&gt;]))
</code></pre>

<p>and <code>codes.first()</code>:</p>

<pre><code>(u'3768915289', (5657, 7810))
</code></pre>

<p>How can I <a href="http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html?highlight=kmeans#pyspark.mllib.clustering.KMeansModel.computeCost" rel="nofollow">computeCost()</a> of this KMeans model?</p>

<hr>

<p>After reading <a href="https://github.com/yahoo/lopq/blob/master/spark/train_model.py" rel="nofollow">train_model.py</a>, I am trying like this:</p>

<pre><code>In [23]: from pyspark.mllib.clustering import KMeans, KMeansModel
In [24]: Cs = model.Cs     # centroids
In [25]: model = KMeansModel(Cs[0]) # I am very positive this line is good
In [26]: costs = d.map(lambda x: model.computeCost(x[1]))
In [27]: costs.first()
</code></pre>

<p>but I get this error:</p>

<pre><code>AttributeError: 'numpy.ndarray' object has no attribute 'map'
</code></pre>

<p>which means that Spark tries to use <code>map()</code> under the hood for <code>x[1]</code>...</p>

<hr>

<p>which means that it expects an RDD!!! But of which form?</p>

<p>I am trying now with:</p>

<pre><code>d = d.map(lambda x: x[1])
d.first()
array([  7.17036494e+01,   1.07987890e+01, ...])
costs = model.computeCost(d)
</code></pre>

<p>and I don't get the error:</p>

<pre><code>16/08/30 00:39:21 WARN TaskSetManager: Lost task 821.0 in stage 40.0 : java.lang.IllegalArgumentException: requirement failed
    at scala.Predef$.require(Predef.scala:221)
    at org.apache.spark.mllib.util.MLUtils$.fastSquaredDistance(MLUtils.scala:330)
    at org.apache.spark.mllib.clustering.KMeans$.fastSquaredDistance(KMeans.scala:595)
    at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:569)
    at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:563)
    at scala.collection.mutable.ArraySeq.foreach(ArraySeq.scala:73)
    at org.apache.spark.mllib.clustering.KMeans$.findClosest(KMeans.scala:563)
    at org.apache.spark.mllib.clustering.KMeans$.pointCost(KMeans.scala:586)
    at org.apache.spark.mllib.clustering.KMeansModel$$anonfun$computeCost$1.apply(KMeansModel.scala:88)
    at org.apache.spark.mllib.clustering.KMeansModel$$anonfun$computeCost$1.apply(KMeansModel.scala:88)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:144)
    at scala.collection.AbstractIterator.foldLeft(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.fold(TraversableOnce.scala:199)
    at scala.collection.AbstractIterator.fold(Iterator.scala:1157)
    at org.apache.spark.rdd.RDD$$anonfun$fold$1$$anonfun$19.apply(RDD.scala:1086)
    at org.apache.spark.rdd.RDD$$anonfun$fold$1$$anonfun$19.apply(RDD.scala:1086)
    at org.apache.spark.SparkContext$$anonfun$36.apply(SparkContext.scala:1951)
    at org.apache.spark.SparkContext$$anonfun$36.apply(SparkContext.scala:1951)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
&lt;ipython-input-44-6223595c8b5f&gt; in &lt;module&gt;()
----&gt; 1 costs = model.computeCost(d)

/home/gs/spark/current/python/pyspark/mllib/clustering.py in computeCost(self, rdd)
    140         """
    141         cost = callMLlibFunc("computeCostKmeansModel", rdd.map(_convert_to_vector),
--&gt; 142                              [_convert_to_vector(c) for c in self.centers])
    143         return cost
    144 

/home/gs/spark/current/python/pyspark/mllib/common.py in callMLlibFunc(name, *args)
    128     sc = SparkContext.getOrCreate()
    129     api = getattr(sc._jvm.PythonMLLibAPI(), name)
--&gt; 130     return callJavaFunc(sc, api, *args)
    131 
    132 

/home/gs/spark/current/python/pyspark/mllib/common.py in callJavaFunc(sc, func, *args)
    121     """ Call Java Function """
    122     args = [_py2java(sc, a) for a in args]
--&gt; 123     return _java2py(sc, func(*args))
    124 
    125 

/home/gs/spark/current/python/lib/py4j-0.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
    811         answer = self.gateway_client.send_command(command)
    812         return_value = get_return_value(
--&gt; 813             answer, self.gateway_client, self.target_id, self.name)
    814 
    815         for temp_arg in temp_args:

/home/gs/spark/current/python/pyspark/sql/utils.py in deco(*a, **kw)
     43     def deco(*a, **kw):
     44         try:
---&gt; 45             return f(*a, **kw)
     46         except py4j.protocol.Py4JJavaError as e:
     47             s = e.java_exception.toString()

/home/gs/spark/current/python/lib/py4j-0.9-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    306                 raise Py4JJavaError(
    307                     "An error occurred while calling {0}{1}{2}.\n".
--&gt; 308                     format(target_id, ".", name), value)
    309             else:
    310                 raise Py4JError(

Py4JJavaError: An error occurred while calling o25177.computeCostKmeansModel.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 821 in stage 40.0 failed 4 times, most recent failure: Lost task 821.3 in stage 40.0: java.lang.IllegalArgumentException: requirement failed
    at scala.Predef$.require(Predef.scala:221)
    at org.apache.spark.mllib.util.MLUtils$.fastSquaredDistance(MLUtils.scala:330)
    at org.apache.spark.mllib.clustering.KMeans$.fastSquaredDistance(KMeans.scala:595)
    at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:569)
    at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:563)
    at scala.collection.mutable.ArraySeq.foreach(ArraySeq.scala:73)
    at org.apache.spark.mllib.clustering.KMeans$.findClosest(KMeans.scala:563)
    at org.apache.spark.mllib.clustering.KMeans$.pointCost(KMeans.scala:586)
    at org.apache.spark.mllib.clustering.KMeansModel$$anonfun$computeCost$1.apply(KMeansModel.scala:88)
    at org.apache.spark.mllib.clustering.KMeansModel$$anonfun$computeCost$1.apply(KMeansModel.scala:88)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:144)
    at scala.collection.AbstractIterator.foldLeft(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.fold(TraversableOnce.scala:199)
    at scala.collection.AbstractIterator.fold(Iterator.scala:1157)
    at org.apache.spark.rdd.RDD$$anonfun$fold$1$$anonfun$19.apply(RDD.scala:1086)
    at org.apache.spark.rdd.RDD$$anonfun$fold$1$$anonfun$19.apply(RDD.scala:1086)
    at org.apache.spark.SparkContext$$anonfun$36.apply(SparkContext.scala:1951)
    at org.apache.spark.SparkContext$$anonfun$36.apply(SparkContext.scala:1951)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
</code></pre>

<hr>

<p>Edit:</p>

<pre><code>split_vecs = d.map(lambda x: np.split(x[1], 2))
</code></pre>

<p>seems to be a good step, since the centroids are of 64 dimensions.</p>

<pre><code>model.computeCost((d.map(lambda x: x[1])).first())
</code></pre>

<p>gives this error: <code>AttributeError: 'numpy.ndarray' object has no attribute 'map'</code>.</p>

<p>Apparently from the <a href="https://spark.apache.org/docs/latest/mllib-clustering.html" rel="nofollow">documentation</a> I've read, you have to:</p>

<ol>
<li><p>Create a model maybe by <a href="http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html?highlight=kmeans#pyspark.mllib.clustering.KMeansModel.load" rel="nofollow">reading</a> a previously <a href="http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html?highlight=kmeans#pyspark.mllib.clustering.KMeansModel.save" rel="nofollow">saved </a> model, or by fitting a new model.</p></li>
<li><p>After obtaining that <a href="http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html?highlight=kmeans#pyspark.mllib.clustering.KMeansModel" rel="nofollow">model</a> you can use its method <a href="http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html?highlight=kmeans#pyspark.mllib.clustering.KMeansModel.clusterCenters" rel="nofollow">computeCost</a>, which needs a well formatted <code>RDD</code> to output something useful. </p></li>
</ol>

<p>Thus, if I assume that your variable <code>model</code> is a <code>KMeansModel</code> and the data stored in the variable <code>d</code> has the expected representation, then you should be able to run the following code:</p>

<pre><code>model.computeCost(d)
</code></pre>

<hr>

<p>Edit:</p>

<p>You should create an RDD that will contain vectors of the same dimensions as the centroids, and provide that as an input parameter to <code>computeCost()</code>, like this for example:</p>

<pre><code>split_vecs = d.map(lambda x: (x[0], np.split(x[1], 2)))
costs_per_split = [KMeansModel(model.Cs[i]).computeCost(split_vecs.map(lambda x: x[1][i])) for i in range(2)]
</code></pre>

python apache-spark machine-learning distributed-computing k-means