使用loadlabledpoints RDD时出现pyspark错误

Question

我正在使用 pyspark

我读取了一个 libsvm 文件，将其转置，然后再次保存。

我将每个数据行保存为具有稀疏数据的 MLUtils.labeledpoint 对象

我尝试使用 MLUtils.saveaslibsvm，然后使用 MLUtils.loadlibsvm 读取文件，但出现以下错误

ValueError：无法将字符串转换为浮点数：[

在 org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193) 在 org.apache.spark.api.python.PythonRunner$$anon$1.(PythonRDD.scala:234) 在 org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152) 在 org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD$$anonfun$8.apply(RDD.scala:336) 在 org.apache.spark.rdd.RDD$$anonfun$8.apply(RDD.scala:334) 在 org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1055) 在 org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1029) 在 org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:969) 在 org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1029) 在 org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:760) 在 org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:334) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:285) 在 org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:108) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ... 1 更多

我在MLUtils 页面中读到，如果你想使用 loadlabeledpoints，你需要使用 RDD.saveAsTextFile 保存数据，但是当我这样做时，我得到了

17/08/10 16:55:51 WARN TaskSetManager：在阶段 1.0 中丢失任务 1.0（TID 3, 192.168.1.205, 执行者 0): org.apache.spark.SparkException: 不能 从以下位置解析双精度： [ 在 org.apache.spark.mllib.util.NumericParser$.parseDouble(NumericParser.scala:120) 在 org.apache.spark.mllib.util.NumericParser$.parseArray(NumericParser.scala:70) 在 org.apache.spark.mllib.util.NumericParser$.parseTuple(NumericParser.scala:91) 在 org.apache.spark.mllib.util.NumericParser$.parse(NumericParser.scala:41) 在 org.apache.spark.mllib.regression.LabeledPoint$.parse(LabeledPoint.scala:62) 在 org.apache.spark.mllib.util.MLUtils$$anonfun$loadLabeledPoints$1.apply(MLUtils.scala:195) 在 org.apache.spark.mllib.util.MLUtils$$anonfun$loadLabeledPoints$1.apply(MLUtils.scala:195) 在 scala.collection.Iterator$$anon$11.next(Iterator.scala:409) 在 org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.next(SerDeUtil.scala:121) 在 org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.next(SerDeUtil.scala:112) 在 scala.collection.Iterator$class.foreach(Ite​​rator.scala:893) 在 org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.foreach(SerDeUtil.scala:112) 在 scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59) 在 scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104) 在 scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48) 在 scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310) 在 org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.to(SerDeUtil.scala:112) 在 scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302) 在 org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.toBuffer(SerDeUtil.scala:112) 在 scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289) 在 org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.toArray(SerDeUtil.scala:112) 在 org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:936) 在 org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:936) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2062) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2062) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:108) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 在 java.lang.Thread.run(Thread.java:748) 引起： java.lang.NumberFormatException：对于输入字符串：“[”在 sun.misc.FloatingDecimal.readJavaFormatString(FloatingDecimal.java:2043) 在 sun.misc.FloatingDecimal.parseDouble(FloatingDecimal.java:110) 在 java.lang.Double.parseDouble(Double.java:538) 在 org.apache.spark.mllib.util.NumericParser$.parseDouble(NumericParser.scala:117) ... 30 更多

如何将标记点的 RDD 保存为 libsvm 格式，然后使用 pyspark 从磁盘加载回来？

谢谢

Answer 1

问题在于将 LabledPoints 写入文件没有使用 libsvm 格式，因此很难重新读取它。

我通过在内存中创建标记点来解决它，然后在将其写入文件之前，将其转换为 libsvm 格式字符串，然后将其写入为文本，之后，我能够将其读取为 libsvm 格式

def pointToLibsvmRow(point):
    s = point.features.reshape(2,-1, order="C").transpose().astype("str")
    pairs = [str(int(float(point.label)))] + ["%s:%s" % (str(int(float(a))), b) for a, b in s.tolist()]
    st = " ".join(pairs)
    return st