title: pyspark: "too many values" error after repartitioning
body: <p>I have a DataFrame (that gets converted to RDD) and would like to repartition so that each key (first column) has its own partition. This is what I did:</p>
  
  <pre><code># Repartition to # key partitions and map each row to a partition given their key rank
  my_rdd = df.rdd.partitionBy(len(keys), lambda row: int(row[0]))
  </code></pre>
  
  <p>However, when I try to map it back to DataFrame or save it, then I get this error:</p>
  
  <pre><code>Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File "spark-1.5.1-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
          process()
        File "spark-1.5.1-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/worker.py",     line 106, in process
  serializer.dump_stream(func(split_index, iterator), outfile)
    File "spark-1.5.1-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/serializers.py", line 133, in dump_stream
      for obj in iterator:
    File "spark-1.5.1-bin-hadoop2.6/python/pyspark/rdd.py", line 1703, in add_shuffle_key
      for k, v in iterator:
  ValueError: too many values to unpack
  
          at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
          at org.apache.spark.api.python.PythonRunner$$anon$1.&lt;init&gt;(PythonRDD.scala:207)
          at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
          at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)
          at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
          at org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)
          at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
          at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
          at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
          at org.apache.spark.scheduler.Task.run(Task.scala:88)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
             at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          ... 1 more
  </code></pre>
  
  <p>A bit more testing revealed that even this causes the same error:
      my_rdd = df.rdd.partitionBy(x) # x = can be 5, 100, etc</p>
  
  <p>Have any of you encountered this before. If so how did you solve it?</p>