title: Pyspark count() and collect() do not work
body: <p>I am confused with my situation. I find sequence patterns in pyspark. 
  At the first I have key value RDD like this </p>
  
  <pre><code>p_split.take(2)
  
  [(['A', 'B', 'C', 'D'], u'749'),
   (['O', 'K', 'A'], u'162')]
  </code></pre>
  
  <p>Than I found combinations of string and join them:</p>
  
  <pre><code>def patterns1(text):
      output = [list(combinations(text, i)) for i in range(len(text) + 1)]
      output = output[2:-1]
      paths = []
      for item in output:
          for i in range(len(item)):
              paths.append('-&gt;'.join(item[i]))
      return paths
  
  
  p_patterns = p_split.map(lambda (x,y): (patterns1(x), y))
  </code></pre>
  
  <p>p_patterns.take(2)</p>
  
  <pre><code> [(['A-&gt;B',
     'A-&gt;C'
     'A-&gt;D',
     'B-&gt;C',
     'B-&gt;D',
     ...
    u'749'), .....
  </code></pre>
  
  <p>And with this RDD p_patterns I can not do operations like count() and collect(). With p_split I did this operations succesfully. </p>
  
  <pre><code>p_patterns.count()
  
      ---------------------------------------------------------------------------
  Py4JJavaError                             Traceback (most recent call last)
  &lt;ipython-input-14-75eb19776fa7&gt; in &lt;module&gt;()
  ----&gt; 1 p_patterns.count()
  
  /usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py in count(self)
      930         3
      931         """
  --&gt; 932         return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
      933 
      934     def stats(self):
  
  /usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py in sum(self)
      921         6.0
      922         """
  --&gt; 923         return self.mapPartitions(lambda x: [sum(x)]).reduce(operator.add)
      924 
      925     def count(self):
  
  /usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py in reduce(self, f)
      737             yield reduce(f, iterator, initial)
      738 
  --&gt; 739         vals = self.mapPartitions(func).collect()
      740         if vals:
      741             return reduce(f, vals)
  
  /usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py in collect(self)
      711         """
      712         with SCCallSiteSync(self.context) as css:
  --&gt; 713             port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
      714         return list(_load_from_socket(port, self._jrdd_deserializer))
      715 
  
  /usr/local/bin/spark-1.3.1-bin-hadoop2.6/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,
  --&gt; 538                 self.target_id, self.name)
      539 
      540         for temp_arg in temp_args:
  
  /usr/local/bin/spark-1.3.1-bin-hadoop2.6/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'.
  --&gt; 300                     format(target_id, '.', name), value)
      301             else:
      302                 raise Py4JError(
  
  Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 8.0 failed 1 times, most recent failure: Lost task 0.0 in stage 8.0 (TID 8, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/worker.py", line 101, in main
      process()
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/worker.py", line 96, in process
      serializer.dump_stream(func(split_index, iterator), outfile)
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py", line 2252, in pipeline_func
      return func(split, prev_func(split, iterator))
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py", line 2252, in pipeline_func
      return func(split, prev_func(split, iterator))
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py", line 2252, in pipeline_func
      return func(split, prev_func(split, iterator))
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py", line 282, in func
      return f(iterator)
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py", line 932, in &lt;lambda&gt;
      return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
    File "/usr/local/bin/spark-1.3.1-bin-hadoop2.6/python/pyspark/rdd.py", line 932, in &lt;genexpr&gt;
      return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
    File "&lt;ipython-input-12-0e1339e78f5c&gt;", line 1, in &lt;lambda&gt;
    File "&lt;ipython-input-11-b71a29b24fa7&gt;", line 7, in patterns1
  MemoryError
  
      at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:135)
      at org.apache.spark.api.python.PythonRDD$$anon$1.&lt;init&gt;(PythonRDD.scala:176)
      at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:94)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
      at org.apache.spark.scheduler.Task.run(Task.scala:64)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
      at java.lang.Thread.run(Thread.java:745)
  
  Driver stacktrace:
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1204)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1193)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1192)
      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:1192)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)
      at scala.Option.foreach(Option.scala:236)
      at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:693)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1393)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1354)
      at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
  </code></pre>
  
  <p>What is my mistake?</p>