How to do feature selection and reduction on a LIBSVM file in Spark using Python?
Question: I have a couple of LIBSVM files with which I have to implement clustering in
spark using python. The file has **space** as the delimiter and the first
column represents the type [ 1 or -1] and the rest all are the features which
are in the format [1:2.566]. There are a lot of columns like this and I would
like to perform a feature selection over this [preferably implement the
ChiSquareTest model] and then use PCA or SVD to perform a feature reduction
process. But, I could not find a decent tutorial for python in spark to
implement these processes.

I found a [link](https://spark.apache.org/docs/1.2.0/mllib-statistics.html)
online that had a sample script to implement Chisqtest in python. I used the
same logic to implement the model and I could not get it done. Under the
Hypothesis testing division in that link, the code parallelizes the
RDD[LabeledPoint] before passing to the ChiSqTest model. I tried the same
logic in different manner and I got different errors.

    
    
    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))
    

This gave me an error stating **TypeError: float() argument must be a string
or a number**

Then, I normalized the data using Normalizer() and did the same thing and got
the same error. So, I wrote a function that returns a labeledpoint

    
    
    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)
    

This gave me an error stating **Pipeline RDD is not iterable**. I tried
several other methods and everything ended up in an error. Could someone
please tell me where I am going wrong? And, for the feature reduction process
using PCA or SVD, I could not find a sample script in python. Any inputs for
that would be very helpful for me.

Stack Trace:

    
    
    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)
    


Answer: `MLUtils.loadLibSVMFile` returns `RDD[LabeledPoint]` so you can pass an output
directly to the `Statistics.chiSqTest`. Using example data:

    
    
    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]
    

