[ https://issues.apache.org/jira/browse/SPARK-18838?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16137365#comment-16137365 ]
Marcelo Vanzin commented on SPARK-18838: ---------------------------------------- [~milesc] after cleaning up a lot of noise from your logs, I could see a bunch of errors, but I can't see how any of them are connected to the listener bus. They seem to be more related to the way Spark is handling those errors and making things worse, instead of correctly handling them. I don't doubt that using the "block the listener bus" patch helps, but it probably helps because of what I said before: that will have an effect on other parts of Spark, and will probably make your application run slower, and will mask these issues. So while that behavior may be better for you, it doesn't help us solve the underlying issues in the code, and Spark running more slowly is hardly something we want. Here are a few choice errors from your log. Shuffle files disappearing. There are quite a few of these. {noformat} 2017-08-15 17:52:17,033 INFO org.apache.spark.scheduler.DAGScheduler: ShuffleMapStage 80 (keyBy at MergeSourcedPapers.scala:94) failed in 266.509 s due to org.apache.spark.shuffle.FetchFailedException: java.io.FileNotFoundException: /mnt/yarn/usercache/milesc/appcache/application_1502651039737_0038/blockmgr-9c34afa1-8738-4003-b532-dc85e04cfb45/26/shuffle_4_4428_0.index (No such file or directory) at java.io.FileInputStream.open0(Native Method) at java.io.FileInputStream.open(FileInputStream.java:195) at java.io.FileInputStream.<init>(FileInputStream.java:138) at org.apache.spark.shuffle.IndexShuffleBlockResolver.getBlockData(IndexShuffleBlockResolver.scala:199) at org.apache.spark.storage.BlockManager.getBlockData(BlockManager.scala:305) at org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:60) {noformat} More missing shuffle data: {noformat} 2017-08-15 17:52:18,120 INFO org.apache.spark.MapOutputTrackerMaster: Size of output statuses for shuffle 3 is 118757 bytes 2017-08-15 17:52:18,121 INFO org.apache.spark.MapOutputTrackerMaster: Epoch changed, not caching! org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 3 at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:697) at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:693) at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:778) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186) at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:777) at org.apache.spark.MapOutputTracker$.org$apache$spark$MapOutputTracker$$convertMapStatuses(MapOutputTracker.sca {noformat} Executors apparently dying: {noformat} 2017-08-15 17:52:19,508 WARN org.apache.spark.network.server.TransportChannelHandler: Exception in connection from /10.12.1.121:51766 java.io.IOException: Connection reset by peer at sun.nio.ch.FileDispatcherImpl.read0(Native Method) at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) at sun.nio.ch.IOUtil.read(IOUtil.java:192) at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380) at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:221) at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:899) {noformat} Executor apparently dying, another type of error: {noformat} 2017-08-15 17:52:26,774 WARN org.apache.spark.storage.BlockManagerMaster: Failed to remove broadcast 56 with removeFromMaster = true - Failed to send RPC 4932545731330680871 to /10.12.1.121:51764: java.nio.channels.ClosedChannelException java.io.IOException: Failed to send RPC 4932545731330680871 to /10.12.1.121:51764: java.nio.channels.ClosedChannelException at org.apache.spark.network.client.TransportClient$3.operationComplete(TransportClient.java:249) at org.apache.spark.network.client.TransportClient$3.operationComplete(TransportClient.java:233) at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:514) at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:488) at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:427) at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:129) {noformat} Another type of executor going away message: {noformat} 2017-08-15 17:52:32,011 INFO org.apache.spark.scheduler.DAGScheduler: Shuffle files lost for executor: 21 (epoch 122) org.apache.spark.shuffle.FetchFailedException: Failed to connect to ip-10-12-1-121.us-west-2.compute.internal/10.12.1.121:33269 at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:361) at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:336) at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:54) at scala.collection.Iterator$$anon$11.next(Iterator.scala:370) ... Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: ip-10-12-1-121.us-west-2.compute.internal/10.12.1.121:33269 at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) {noformat} YARN scheduler getting backed up: {noformat} 2017-08-15 17:56:19,816 ERROR org.apache.spark.scheduler.cluster.YarnClusterScheduler: Lost executor 23 on ip-10-12-13-61.us-west-2.compute.internal: Slave lost 2017-08-15 17:56:19,863 WARN org.apache.spark.scheduler.cluster.YarnSchedulerBackend$YarnSchedulerEndpoint: Attempted to get executor loss reason for executor id 21 at RPC address 10.12.1.121:51766, but got no response. Marking as slave lost. org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in 240 seconds. This timeout is controlled by spark.network.timeout at org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48) at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63) {noformat} Weird network issues: {noformat} 2017-08-15 19:45:14,126 INFO org.apache.spark.ContextCleaner: Cleaned accumulator 13006962 org.apache.spark.shuffle.FetchFailedException: Failed to connect to ip-10-12-12-210.us-west-2.compute.internal/10.12.12.210:38223 at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:361) at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:336) at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:54) at scala.collection.Iterator$$anon$11.next(Iterator.scala:370) at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:396) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369) at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32) at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39) ... Caused by: io.netty.channel.AbstractChannel$AnnotatedNoRouteToHostException: No route to host: ip-10-12-12-210.us-west-2.compute.internal/10.12.12.210:38223 at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) {noformat} And it all ends with Spark apparently endlessly trying to distribute a broadcast variable to the same executor over and over again, without making any progress. Something that might be understandable with a longer look at the logs (and if the executor logs were also available). So there's a whole bunch of things going on, some of which I've seen before and might even be fixed already, but I can't really see anything that is directly related to the listener bus being backed up. > High latency of event processing for large jobs > ----------------------------------------------- > > Key: SPARK-18838 > URL: https://issues.apache.org/jira/browse/SPARK-18838 > Project: Spark > Issue Type: Improvement > Affects Versions: 2.0.0 > Reporter: Sital Kedia > Attachments: perfResults.pdf, SparkListernerComputeTime.xlsx > > > Currently we are observing the issue of very high event processing delay in > driver's `ListenerBus` for large jobs with many tasks. Many critical > component of the scheduler like `ExecutorAllocationManager`, > `HeartbeatReceiver` depend on the `ListenerBus` events and this delay might > hurt the job performance significantly or even fail the job. For example, a > significant delay in receiving the `SparkListenerTaskStart` might cause > `ExecutorAllocationManager` manager to mistakenly remove an executor which is > not idle. > The problem is that the event processor in `ListenerBus` is a single thread > which loops through all the Listeners for each event and processes each event > synchronously > https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala#L94. > This single threaded processor often becomes the bottleneck for large jobs. > Also, if one of the Listener is very slow, all the listeners will pay the > price of delay incurred by the slow listener. In addition to that a slow > listener can cause events to be dropped from the event queue which might be > fatal to the job. > To solve the above problems, we propose to get rid of the event queue and the > single threaded event processor. Instead each listener will have its own > dedicate single threaded executor service . When ever an event is posted, it > will be submitted to executor service of all the listeners. The Single > threaded executor service will guarantee in order processing of the events > per listener. The queue used for the executor service will be bounded to > guarantee we do not grow the memory indefinitely. The downside of this > approach is separate event queue per listener will increase the driver memory > footprint. -- This message was sent by Atlassian JIRA (v6.4.14#64029) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org