Uploaded image for project: 'Spark'
  1. Spark
  2. SPARK-4841

Batch serializer bug in PySpark's RDD.zip

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: Blocker
    • Resolution: Fixed
    • Affects Version/s: 1.2.0
    • Fix Version/s: 1.2.1, 1.3.0
    • Component/s: PySpark
    • Labels:
      None

      Description

      t = sc.textFile("README.md")
      t.zip(t).count()
      
      Py4JJavaError                             Traceback (most recent call last)
      <ipython-input-6-60fdeb8339fd> in <module>()
      ----> 1 readme.zip(readme).count()
      
      /Users/meng/src/spark/python/pyspark/rdd.pyc in count(self)
          817         3
          818         """
      --> 819         return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
          820
          821     def stats(self):
      
      /Users/meng/src/spark/python/pyspark/rdd.pyc in sum(self)
          808         6.0
          809         """
      --> 810         return self.mapPartitions(lambda x: [sum(x)]).reduce(operator.add)
          811
          812     def count(self):
      
      /Users/meng/src/spark/python/pyspark/rdd.pyc in reduce(self, f)
          713             yield reduce(f, iterator, initial)
          714
      --> 715         vals = self.mapPartitions(func).collect()
          716         if vals:
          717             return reduce(f, vals)
      
      /Users/meng/src/spark/python/pyspark/rdd.pyc in collect(self)
          674         """
          675         with SCCallSiteSync(self.context) as css:
      --> 676             bytesInJava = self._jrdd.collect().iterator()
          677         return list(self._collect_iterator_through_file(bytesInJava))
          678
      
      /Users/meng/src/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:
      
      /Users/meng/src/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 o69.collect.
      : 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): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
        File "/Users/meng/src/spark/python/pyspark/worker.py", line 107, in main
          process()
        File "/Users/meng/src/spark/python/pyspark/worker.py", line 98, in process
          serializer.dump_stream(func(split_index, iterator), outfile)
        File "/Users/meng/src/spark/python/pyspark/serializers.py", line 198, in dump_stream
          self.serializer.dump_stream(self._batched(iterator), stream)
        File "/Users/meng/src/spark/python/pyspark/serializers.py", line 81, in dump_stream
          raise NotImplementedError
      NotImplementedError
      
      	at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)
      	at org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)
      	at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)
      	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
      	at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
      	at org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:88)
      	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
      	at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
      	at org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply$mcV$sp(PythonRDD.scala:242)
      	at org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply(PythonRDD.scala:204)
      	at org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply(PythonRDD.scala:204)
      	at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1460)
      	at org.apache.spark.api.python.PythonRDD$WriterThread.run(PythonRDD.scala:203)
      
      Driver stacktrace:
      	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)
      	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)
      	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)
      	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:1202)
      	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)
      	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)
      	at scala.Option.foreach(Option.scala:236)
      	at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)
      	at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)
      	at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
      	at org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)
      	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
      	at akka.actor.ActorCell.invoke(ActorCell.scala:487)
      	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
      	at akka.dispatch.Mailbox.run(Mailbox.scala:220)
      	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
      	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
      	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
      	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
      	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                davies Davies Liu
                Reporter:
                mengxr Xiangrui Meng
              • Votes:
                0 Vote for this issue
                Watchers:
                4 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: