Not use LabeledPoint

Author: yu-iskw

mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala

@@ -506,7 +506,7 @@ private[python] class PythonMLLibAPI extends Serializable {
    * Java stub for Python mllib LDA.run()
    */
   def trainLDAModel(
-      data: JavaRDD[LabeledPoint],
+      data: JavaRDD[java.util.List[Any]],
       k: Int,
       maxIterations: Int,
       docConcentration: Double,
@@ -524,11 +524,14 @@ private[python] class PythonMLLibAPI extends Serializable {
 
     if (seed != null) algo.setSeed(seed)
 
-    try {
-      algo.run(data.rdd.map(x => (x.label.toLong, x.features)))
-    } finally {
-      data.rdd.unpersist(blocking = false)
+    val documents = data.rdd.map(_.asScala.toArray).map { r =>
+      r(0).getClass.getSimpleName match {
+        case "Integer" =>  (r(0).asInstanceOf[java.lang.Integer].toLong, r(1).asInstanceOf[Vector])
+        case "Long" =>  (r(0).asInstanceOf[java.lang.Long].toLong, r(1).asInstanceOf[Vector])
+        case _ => throw new IllegalArgumentException("input values contains invalid type value.")
+      }
     }
+    algo.run(documents)
   }
 
 

python/pyspark/mllib/clustering.py

@@ -593,8 +593,8 @@ class LDAModel(JavaModelWrapper):
     >>> from collections import namedtuple
     >>> from numpy.testing import assert_almost_equal
     >>> data = [
-    ...     LabeledPoint(1, [0.0, 1.0]),
-    ...     LabeledPoint(2, [1.0, 0.0]),
+    ...     [1, Vectors.dense([0.0, 1.0])],
+    ...     [2, SparseVector(2, {0: 1.0})],
     ... ]
     >>> rdd =  sc.parallelize(data)
     >>> model = LDA.train(rdd, k=2)