A JVM can easily be limited in how much memory it uses with the -Xmx
parameter, but Python doesn't have memory limits built in in such a
first-class way. Maybe the memory limits aren't making it to the python
executors.
What was your SPARK_MEM setting? The JVM below seems to be using 603201
(pages?) and the 3 large python processes each are using ~1800000 (pages?).
I'm unsure the units that the OOM killer's RSS column is in. Could be
either pages (4kb each) or bytes.
Apr 8 11:19:19 bennett kernel: [86368.978326] [ 2348] 1002 2348 12573
2102 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978329] [ 2349] 1002 2349 12573
2101 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978332] [ 2350] 1002 2350 12573
2101 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978336] [ 5115] 1002 5115 12571
2101 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978339] [ 5116] 1002 5116 12571
2101 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978341] [ 5117] 1002 5117 12571
2101 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978344] [ 7725] 1002 7725 12570
2098 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978347] [ 7726] 1002 7726 12570
2098 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978350] [ 7727] 1002 7727 12570
2098 22 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978353] [10324] 1002 10324 12570
2098 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978356] [10325] 1002 10325 12570
2098 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978359] [10326] 1002 10326 12570
2098 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978362] [12668] 1002 12668 603201
47932 190 0 0 java
Apr 8 11:19:19 bennett kernel: [86368.978366] [13295] 1002 13295 12570
2100 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978368] [13296] 1002 13296 12570
2100 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978371] [13297] 1002 13297 12570
2100 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978375] [15192] 1002 15192 12570
2098 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978377] [15193] 1002 15193 12570
2098 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978379] [15195] 1002 15195 12570
2098 23 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978381] [15198] 1002 15198 1845471
1818463 3573 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978383] [15200] 1002 15200 1710479
1686492 3316 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978384] [15201] 1002 15201 1788470
1762344 3463 0 0 python
Apr 8 11:19:19 bennett kernel: [86368.978386] Out of memory: Kill process
15198 (python) score 221 or sacrifice child
Apr 8 11:19:19 bennett kernel: [86368.978389] Killed process 15198
(python) total-vm:7381884kB, anon-rss:7273852kB, file-rss:0kB
On Tue, Apr 8, 2014 at 2:56 PM, Brad Miller <bmill...@eecs.berkeley.edu>wrote:
> Hi All,
>
> I poked around a bit more to (1) confirm my suspicions that the crash
> was related to memory consumption and (2) figure out why there is no
> error shown in 12_stderr, the spark executor log file from the
> executors on bennett.research.intel.research.net.
>
> The syslog file (from /var/log/syslog on bennett, attached) shows that
> the machine ran out of memory, the memory was mostly consumed by 1
> java process and 3 python processes (I am running pyspark with 3 cores
> per machine), and then the kernel began killing java and python
> processes to ease memory pressure. It seems likely that these
> processes were the spark processes, and there's no errors recorded in
> 12_stderr because the process was killed by the OS (rather than
> experiencing an unhandled "cannot allocate memory" exception).
>
> I'm a little confused how Spark could consume so much memory during
> the reduce phase of the shuffle. Shouldn't Spark remain within the
> SPARK_MEM limitations on memory consumption, and spill to disk in the
> event that there isn't enough memory?
>
> -Brad
>
>
> On Tue, Apr 8, 2014 at 12:50 PM, Brad Miller <bmill...@eecs.berkeley.edu>
> wrote:
> > Hi Patrick,
> >
> >> The shuffle data is written through the buffer cache of the operating
> >> system, so you would expect the files to show up there immediately and
> >> probably to show up as being their full size when you do "ls". In
> reality
> >> though these are likely residing in the OS cache and not on disk.
> >
> > I see. Perhaps the memory consumption is related to this?
> >
> >> Could you paste the error here?
> >
> > While I have definitely seen "cannot allocate memory" errors while
> > trying to do this, I am unable to reproduce one now. Instead, I am
> > able to produce "most recent failure: unknown" (see full error
> > displayed in my iPython session below). Initially, I assumed there
> > was some sort of non-determinism which caused the error to
> > occasionally be "unknown", but now I realize that it may have been a
> > consistent change which occurred when I updated to the latest
> > brach-0.9 (previously I was running a version I pulled around March
> > 10th).
> >
> > Py4JJavaError: An error occurred while calling o274.collect.
> > : org.apache.spark.SparkException: Job aborted: Task 2.0:29 failed 4
> > times (most recent failure: unknown)
> > at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> > at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> > 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.org
> $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> > at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > at scala.Option.foreach(Option.scala:236)
> > at
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> > at
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> > 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)
> >
> >> From the logs it looks like your executor has died. Would you be able to
> >> paste the log from the executor with the exact failure? It would show
> up in
> >> the /work directory inside of spark's directory on the cluster.
> >
> > I've attached the logging output from the driver from when I re-ran
> > the join operation this morning. It seems that specific, individual
> > workers (or perhaps executors is the right term?) begin to die and
> > then are re-launched by the master/driver. When I examine the
> > app-xxx...xxx folder corresponding to this job on the first worker to
> > fail (bennett.research.intel-research.net), there are several numbered
> > folders inside (12, 15, 21, 22, 23) which seem to correspond to each
> > invocation of the executor as recorded in the driver log. stdout is
> > consistently empty, and stderr is not. I have attached all of these
> > logs as <executor_id>_stderr.
> >
> > Surprisingly, 12_stderr does not record any sort of error, although
> > 15_stderr, 21_stderr, and 22_stderr do. These errors are all of the
> > form:
> >
> > 14/04/08 11:19:42 ERROR Executor: Uncaught exception in thread
> > Thread[stdin writer for python,5,main]
> > org.apache.spark.FetchFailedException: Fetch failed: null 0 -1 44
> > at
> org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$1.apply(MapOutputTracker.scala:316)
> > at
> org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$1.apply(MapOutputTracker.scala:314)
> > at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> > at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> > at
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> > at
> scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
> > at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
> > at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108)
> > at
> org.apache.spark.MapOutputTracker$.org$apache$spark$MapOutputTracker$$convertMapStatuses(MapOutputTracker.scala:313)
> > at
> org.apache.spark.MapOutputTracker.getServerStatuses(MapOutputTracker.scala:139)
> > at
> org.apache.spark.BlockStoreShuffleFetcher.fetch(BlockStoreShuffleFetcher.scala:43)
> > at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:61)
> > at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241)
> > at org.apache.spark.rdd.RDD.iterator(RDD.scala:232)
> > at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)
> > at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241)
> > at org.apache.spark.rdd.RDD.iterator(RDD.scala:232)
> > at
> org.apache.spark.api.python.PythonRDD$$anon$2.run(PythonRDD.scala:85)
> > Caused by: java.lang.Exception: Missing an output location for shuffle 0
> > at
> org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$1.apply(MapOutputTracker.scala:317)
> > ... 17 more
> >
> > Note that in "Fetch failed: null 0 -1 44" the last numeral varies in
> > each of the errors. I have also attached an executor stderr trace
> > from malkin.research.intel-research.net, which one of several
> > executors to crash after bennett. Notice that this trace does contain
> > errors, although they seem to be saying that there's trouble
> > connecting to bennett.
> >
> > If anyone would like additional information, please let me know.
> >
> > best,
> > -Brad
> >
> > On Mon, Apr 7, 2014 at 9:40 PM, Patrick Wendell <pwend...@gmail.com>
> wrote:
> >>
> >>
> >>
> >> On Mon, Apr 7, 2014 at 7:37 PM, Brad Miller <bmill...@eecs.berkeley.edu
> >
> >> wrote:
> >>>
> >>> I am running the latest version of PySpark branch-0.9 and having some
> >>> trouble with join.
> >>>
> >>> One RDD is about 100G (25GB compressed and serialized in memory) with
> >>> 130K records, the other RDD is about 10G (2.5G compressed and
> >>> serialized in memory) with 330K records. I load each RDD from HDFS,
> >>> invoke keyBy to key each record, and then attempt to join the RDDs.
> >>>
> >>> The join consistently crashes at the beginning of the reduce phase.
> >>> Note that when joining the 10G RDD to itself there is no problem.
> >>>
> >>> Prior to the crash, several suspicious things happen:
> >>>
> >>> -All map output files from the map phase of the join are written to
> >>> spark.local.dir, even though there should be plenty of memory left to
> >>> contain the map output. I am reasonably sure *all* map outputs are
> >>> written to disk because the size of the map output spill directory
> >>> matches the size of the shuffle write (as displayed in the user
> >>> interface) for each machine.
> >>
> >>
> >> The shuffle data is written through the buffer cache of the operating
> >> system, so you would expect the files to show up there immediately and
> >> probably to show up as being their full size when you do "ls". In
> reality
> >> though these are likely residing in the OS cache and not on disk.
> >>
> >>>
> >>> -In the beginning of the reduce phase of the join, memory consumption
> >>> on each work spikes and each machine runs out of memory (as evidenced
> >>> by a "Cannon allocate memory" exception in Java). This is
> >>> particularly surprising since each machine has 30G of ram and each
> >>> spark worker has only been allowed 10G.
> >>
> >>
> >> Could you paste the error here?
> >>
> >>>
> >>> -In the web UI both the "Shuffle Spill (Memory)" and "Shuffle Spill
> >>> (Disk)" fields for each machine remain at 0.0 despite shuffle files
> >>> being written into spark.local.dir.
> >>
> >>
> >> Shuffle spill is different than the shuffle files written to
> >> spark.local.dir. Shuffle spilling is for aggregations that occur on the
> >> reduce side of the shuffle. A join like this might not see any spilling.
> >>
> >>
> >>
> >> From the logs it looks like your executor has died. Would you be able to
> >> paste the log from the executor with the exact failure? It would show
> up in
> >> the /work directory inside of spark's directory on the cluster.
>
