Github user giwa commented on the pull request:
https://github.com/apache/spark/pull/2538#issuecomment-57148324
@davies
Python API for Streaming becomes much nicer. You are awesome! Tests have
been failed so far. But it looks like other issue, I think.
Thanks to you, callback server is terminated properly even after error is
occurred, as far as I know.
I got errors when I executed run-tests. I pasted log below. This error is
not always happened. When I run only TestWindowFunctions by ./bin/pyspark
python/pyspark/streaming/tests.py TestWindowFunctions, I have not seen this
error.
My environment:
Mac OS 10.10
Python 2.6.9
Java 1.7.0_65
Scala 2.10.3
```
Traceback (most recent call last):
File "/Users/ken/spark/davies/spark/python/pyspark/streaming/util.py",
line 39, in call
r = self.func(rdd, other, milliseconds)
File "/Users/ken/spark/davies/spark/python/pyspark/streaming/dstream.py",
line 145, in <lambda>
jfunc = RDDFunction(self.ctx, lambda a, _, t: func(a, t),
self._jrdd_deserializer)
File "/Users/ken/spark/davies/spark/python/pyspark/streaming/dstream.py",
line 199, in get_output
r = rdd.collect()
File "/Users/ken/spark/davies/spark/python/pyspark/rdd.py", line 656, in
collect
bytesInJava = self._jrdd.collect().iterator()
File
"/Users/ken/spark/davies/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py",
line 538, in __call__
self.target_id, self.name)
File
"/Users/ken/spark/davies/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py",
line 300, in get_return_value
format(target_id, '.', name), value)
Py4JJavaError: An error occurred while calling o7478.collect.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0
in stage 34.0 failed 1 times, most recent failure: Lost task 0.0 in stage 34.0
(TID 190, localhost): java.io.IOException: PARSING_ERROR(2)
org.xerial.snappy.SnappyNative.throw_error(SnappyNative.java:84)
org.xerial.snappy.SnappyNative.uncompressedLength(Native Method)
org.xerial.snappy.Snappy.uncompressedLength(Snappy.java:594)
org.xerial.snappy.SnappyInputStream.readFully(SnappyInputStream.java:125)
org.xerial.snappy.SnappyInputStream.readHeader(SnappyInputStream.java:88)
org.xerial.snappy.SnappyInputStream.<init>(SnappyInputStream.java:58)
org.apache.spark.io.SnappyCompressionCodec.compressedInputStream(CompressionCodec.scala:128)
org.apache.spark.broadcast.TorrentBroadcast$.unBlockifyObject(TorrentBroadcast.scala:216)
org.apache.spark.broadcast.TorrentBroadcast.readObject(TorrentBroadcast.scala:170)
sun.reflect.GeneratedMethodAccessor9.invoke(Unknown Source)
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
java.lang.reflect.Method.invoke(Method.java:606)
java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
java.util.ArrayList.readObject(ArrayList.java:771)
sun.reflect.GeneratedMethodAccessor6.invoke(Unknown Source)
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
java.lang.reflect.Method.invoke(Method.java:606)
java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
scala.collection.immutable.$colon$colon.readObject(List.scala:362)
sun.reflect.GeneratedMethodAccessor4.invoke(Unknown Source)
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
java.lang.reflect.Method.invoke(Method.java:606)
java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:62)
org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:87)
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:60)
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
org.apache.spark.scheduler.Task.run(Task.scala:54)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:182)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
java.lang.Thread.run(Thread.java:745)
Driver stacktrace:
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1190)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1179)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1178)
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:1178)
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.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1396)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
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)
======================================================================
FAIL: test_count_by_window (__main__.TestWindowFunctions)
----------------------------------------------------------------------
Traceback (most recent call last):
File "pyspark/streaming/tests.py", line 284, in test_count_by_window
self._test_func(input, func, expected)
File "pyspark/streaming/tests.py", line 78, in _test_func
self.assertEqual(expected, result)
AssertionError: [[1], [3], [6], [9], [12], [15], [11], [6]] != [[1], [12],
[], [], [], [], [], []]
======================================================================
FAIL: test_count_by_window_large (__main__.TestWindowFunctions)
----------------------------------------------------------------------
Traceback (most recent call last):
File "pyspark/streaming/tests.py", line 293, in test_count_by_window_large
self._test_func(input, func, expected)
File "pyspark/streaming/tests.py", line 78, in _test_func
self.assertEqual(expected, result)
AssertionError: [[1], [3], [6], [10], [15], [20], [18], [15], [11], [6]] !=
[[1]]
======================================================================
FAIL: test_group_by_key_and_window (__main__.TestWindowFunctions)
----------------------------------------------------------------------
Traceback (most recent call last):
File "pyspark/streaming/tests.py", line 303, in
test_group_by_key_and_window
self._test_func(input, func, expected)
File "pyspark/streaming/tests.py", line 78, in _test_func
self.assertEqual(expected, result)
AssertionError: [[('a', [0])], [('a', [0, 1])], [('a', [0, 1, 2])], [('a',
[1, 2, 3])], [('a', [2, 3, 4])], [('a', [3, 4])], [('a', [4])]] != [[('a',
[0])], [('a', [2, 3, 4])], [], [], [], [], []]
----------------------------------------------------------------------
Ran 24 tests in 148.729s
FAILED (failures=3)
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
