python之如何使用 Python 在 Spark 中对 LIBSVM 文件进行特征选择和缩减

我有几个 LIBSVM 文件，我必须使用它们在 spark 中使用 python 实现集群。该文件以空格作为分隔符，第一列表示类型[1或-1]，其余均为[1:2.566]格式的特征。有很多这样的列，我想对此进行特征选择[最好实现 ChiSquareTest 模型]，然后使用 PCA 或 SVD 进行特征缩减过程。但是，我无法在 spark 中找到一个像样的 python 教程来实现这些过程。

我找到了一个 link在线有一个在 python 中实现 Chisqtest 的示例脚本。我使用相同的逻辑来实现模型，但无法完成。在该链接的假设检验部分下，代码在传递给 ChiSqTest 模型之前将 RDD[LabeledPoint] 并行化。我以不同的方式尝试了相同的逻辑，但出现了不同的错误。

data = MLUtils.loadLibSVMFile(sc, "PATH/FILENAME.txt") 
label = data.map(lambda x: x.label) 
features = data.map(lambda x: x.features) 
obs = sc.parallelize(LabeledPoint(label,features)) 

这给了我一个错误，指出 TypeError: float() argument must be a string or a number

然后，我使用 Normalizer() 对数据进行了规范化，并做了同样的事情并得到了同样的错误。所以，我写了一个返回标记点的函数

def parsepoint(line): 
    values = [float(x) for x in line.split(' ')] 
    return sc.parallelize(LabeledPoint(values[0],values[1:])) 
parsedData = data.map(lambda x: parsepoint(x)) 
obs = sc.parallelize(parsedData) 

这给了我一个错误，指出 Pipeline RDD 不可迭代。我尝试了其他几种方法，但一切都以错误告终。有人可以告诉我哪里出错了吗？而且，对于使用 PCA 或 SVD 的特征缩减过程，我找不到 python 中的示例脚本。对此的任何输入都会对我很有帮助。

堆栈跟踪:

Py4JJavaError                             Traceback (most recent call last) 
<ipython-input-1-8d0164c0957d> in <module>() 
  10 sct = SparkContext() 
  11 data = MLUtils.loadLibSVMFile(sct, "PATH") 
  ---> 12 print data.take(1) 
  13 #label = data.map(lambda x: x.label) 
  14 #features = data.map(lambda x: x.features) 
 
  SPARK_HOME\rdd.pyc in take(self, num) 
  1263  
  1264 p = range(partsScanned, min(partsScanned + numPartsToTry,   totalParts)) 
 -> 1265 res = self.context.runJob(self, takeUpToNumLeft, p, True) 
  1266  
  1267 items += res 
 
  SPARK_HOME\context.pyc in runJob(self, rdd, partitionFunc, partitions, allowLocal) 
   879         mappedRDD = rdd.mapPartitions(partitionFunc) 
   880         port = self._jvm.PythonRDD.runJob(self._jsc.sc(),   mappedRDD._jrdd, partitions, 
   --> 881 allowLocal) 
   882         return list(_load_from_socket(port, mappedRDD._jrdd_deserializer)) 
   883  
      SPARK\python\lib\py4j-0.8.2.1-src.zip\py4j\java_gateway.py in __call__(self, *args) 
   536 answer = self.gateway_client.send_command(command) 
   537 return_value = get_return_value(answer, self.gateway_client, 
   --> 538 self.target_id, self.name) 
   539  
   540  for temp_arg in temp_args: 
    SPARK\python\lib\py4j-0.8.2.1-src.zip\py4j\protocol.py in get_return_value(answer, gateway_client, target_id, name) 
   298 raise Py4JJavaError( 
   299 'An error occurred while calling {0}{1}{2}.\n'. 
   --> 300  format(target_id, '.', name), value) 
   301   else: 
   302   raise Py4JError( 
 
  Py4JJavaError: An error occurred while calling       z:org.apache.spark.api.python.PythonRDD.runJob. 
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task   0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0   (TID 2, localhost): java.net.SocketException: Connection reset by peer: socket      write error 
   at java.net.SocketOutputStream.socketWrite0(Native Method) 
   at java.net.SocketOutputStream.socketWrite(Unknown Source) 
   at java.net.SocketOutputStream.write(Unknown Source) 
   at java.io.BufferedOutputStream.write(Unknown Source) 
   at java.io.DataOutputStream.write(Unknown Source) 
   at java.io.FilterOutputStream.write(Unknown Source) 
   at    org.apache.spark.api.python.PythonRDD$.org$apache$spark$api$python$PythonRDD$$wr       ite$1(PythonRDD.scala:413) 
   at   org.apache.spark.api.python.PythonRDD$$anonfun$writeIteratorToStream$1.apply(PythonRDD.scala:425) 
   at  org.apache.spark.api.python.PythonRDD$$anonfun$writeIteratorToStream$1.apply(PythonRDD.scala:425) 
   at scala.collection.Iterator$class.foreach(Iterator.scala:727) 
   at   org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28) 
   at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:425) 
   at org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$3.apply(PythonRD D.scala:248) 
   at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1772) 
   at org.apache.spark.api.python.PythonRDD$WriterThread.run(PythonRDD.scala:208) 
 
  Driver stacktrace: 
   at   org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$ $failJobAndIndependentStages(DAGScheduler.scala:1266) 
  at  org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler .scala:1257) 
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1256) 
  at   scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 
 at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 
  at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1256) 
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAG Scheduler.scala:730) 
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:730) 
  at scala.Option.foreach(Option.scala:236) 
  at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:730) 
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1450) 
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala :1411) 
 at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) 

请您参考如下方法:

MLUtils.loadLibSVMFile 返回 RDD[LabeledPoint]，因此您可以将输出直接传递给 Statistics.chiSqTest。使用示例数据:

from pyspark.mllib.util import MLUtils 
from pyspark.mllib.stat import Statistics 
 
data = MLUtils.loadLibSVMFile(sc, 'data/mllib/sample_libsvm_data.txt') 
chiSqResults = Statistics.chiSqTest(data) 
 
print chiSqResults[-1]