[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15689843#comment-15689843
]
Roy wang commented on SPARK-14560:
--
I found my tasks run into this problem after them was running nearly 2 hours
since one week ago. I am using spark 2.0.2 on hadoop 2.6.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
--
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15690589#comment-15690589
]
Lianhui Wang commented on SPARK-14560:
--
Can you provide some debug log of TaskMemoryManager? I think that can tell who
use the memory.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To u
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15694677#comment-15694677
]
Roy Wang commented on SPARK-14560:
--
UnsafeExternalSorter used a lot of memory but it looked like that
TaskMemoryManager couldn't spill these consumers when acquireExecutionMemory
method was called.
public long acquireExecutionMemory(long required, MemoryConsumer consumer) {
assert(required >= 0);
assert(consumer != null);
MemoryMode mode = consumer.getMode();
// If we are allocating Tungsten pages off-heap and receive a request to
allocate on-heap
// memory here, then it may not make sense to spill since that would only
end up freeing
// off-heap memory. This is subject to change, though, so it may be risky
to make this
// optimization now in case we forget to undo it late when making changes.
synchronized (this) {
long got = memoryManager.acquireExecutionMemory(required, taskAttemptId,
mode);
// Try to release memory from other consumers first, then we can reduce
the frequency of
// spilling, avoid to have too many spilled files.
if (got < required) {
// Call spill() on other consumers to release memory
for (MemoryConsumer c: consumers) {
if (c != consumer && c.getUsed() > 0 && c.getMode() == mode) {
try {
long released = c.spill(required - got, consumer);
if (released > 0) {
logger.info("Task {} released {} from {} for {} by {} mode",
taskAttemptId,
Utils.bytesToString(released), c, consumer, mode.name());
got += memoryManager.acquireExecutionMemory(required - got,
taskAttemptId, mode);
if (got >= required) {
break;
}
}
} catch (IOException e) {
logger.error("error while calling spill() on " + c, e);
throw new OutOfMemoryError("error while calling spill() on " + c
+ " : "
+ e.getMessage());
}
}
}
}
// call spill() on itself
if (got < required) {
try {
long released = consumer.spill(required - got, consumer);
if (released > 0) {
logger.debug("Task {} released {} from itself ({})", taskAttemptId,
Utils.bytesToString(released), consumer);
got += memoryManager.acquireExecutionMemory(required - got,
taskAttemptId, mode);
}
} catch (IOException e) {
logger.error("error while calling spill() on " + consumer, e);
throw new OutOfMemoryError("error while calling spill() on " +
consumer + " : "
+ e.getMessage());
}
}
consumers.add(consumer);
logger.debug("Task {} acquired {} for {}", taskAttemptId,
Utils.bytesToString(got), consumer);
return got;
}
}
Here is the debug log when OOM happened:
16/11/24 06:20:26 INFO TaskMemoryManager: 0 bytes of memory were used by task
72969 but are not associated with specific consumers
16/11/24 06:20:26 INFO TaskMemoryManager: 18911656686 bytes of memory are used
for execution and 1326241517 bytes of memory are used for storage
16/11/24 06:20:26 INFO TaskMemoryManager: Memory used in task 72969
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@45f5aaf1:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@25bc0eb9:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@7ca441ad:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@7c4b6a45:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@152d313e:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@f936c87: 64.0
MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@578f36d: 64.0
MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@4f4ab652:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@6947ad58:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@38eb70df:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter@6c4c0f79:
64.0 MB
16/11/24 06:20:26 INFO TaskMemoryManager: Acquired by
org.apache.sp
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[ https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15666005#comment-15666005 ] KaiXu commented on SPARK-14560: --- spark1.6.2 has the same issue: 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16761.0 in stage 1.0 (TID 17639) in 11657 ms on eurus-node4 (16634/22488) 16/11/15 10:11:15 INFO scheduler.DAGScheduler: Executor lost: 13 (epoch 1) 16/11/15 10:11:15 WARN scheduler.TaskSetManager: Lost task 10002.0 in stage 1.0 (TID 10880, eurus-node3): java.lang.OutOfMemoryError: Unable to acquire 51 bytes of memory, got 0 at org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120) at org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:359) at org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:380) at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237) at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164) 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:89) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16766.0 in stage 1.0 (TID 17644) in 11508 ms on eurus-node4 (16635/22488) 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16574.0 in stage 1.0 (TID 17452) in 21354 ms on eurus-node1 (16636/22488) 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16733.0 in stage 1.0 (TID 17611) in 13122 ms on eurus-node2 (16637/22488) 16/11/15 10:11:15 INFO storage.BlockManagerMasterEndpoint: Trying to remove executor 13 from BlockManagerMaster. 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16735.0 in stage 1.0 (TID 17613) in 12888 ms on eurus-node2 (16638/22488) 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16723.0 in stage 1.0 (TID 17601) in 13572 ms on eurus-node2 (16639/22488) 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16521.0 in stage 1.0 (TID 17399) in 24803 ms on eurus-node4 (16640/22488) 16/11/15 10:11:15 INFO scheduler.TaskSetManager: Finished task 16722.0 in stage 1.0 (TID 17600) in 13621 ms on eurus-node2 (16641/22488) 16/11/15 10:11:15 INFO storage.BlockManagerMasterEndpoint: Removing block manager BlockManagerId(13, eurus-node3, 42404) 16/11/15 10:11:15 WARN scheduler.TaskSetManager: Lost task 16962.0 in stage 1.0 (TID 17840, eurus-node4): FetchFailed(BlockManagerId(13, eurus-node3, 42404), shuffleId=4, mapId=35, reduceId=16962, message= org.apache.spark.shuffle.FetchFailedException: java.io.FileNotFoundException: /mnt/disk1/yarn/nm/usercache/root/appcache/application_147917013_0002/blockmgr-0161eaa3-ea02-40b2-81f7-082606db1271/1b/shuffle_4_35_0.index (No such file or directory) at java.io.FileInputStream.open(Native Method) at java.io.FileInputStream.(FileInputStream.java:146) at org.apache.spark.shuffle.IndexShuffleBlockResolver.getBlockData(IndexShuffleBlockResolver.scala:191) at org.apache.spark.storage.BlockManager.getBlockData(BlockManager.scala:298) at org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:58) at org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:58) 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.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:58) at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:149) at org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:102) at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:104) at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportC
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15670007#comment-15670007
]
Lianhui Wang commented on SPARK-14560:
--
Now this issue is not in Branch-1.6. You can see
https://github.com/apache/spark/pull/13027 for branch-1.6. Thanks.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15823296#comment-15823296
]
Morten Hornbech commented on SPARK-14560:
-
I have also observed this error sporadically on Spark 2.0.2. Does the
spark.shuffle.spill.numElementsForceSpillThreshold=N workaround work on 2.0.2?
Any experience with robust values?
Stack trace:
java.lang.OutOfMemoryError: Unable to acquire 36 bytes of memory, got 0
at
org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:129)
at
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.acquireNewPageIfNecessary(UnsafeExternalSorter.java:377)
at
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.insertRecord(UnsafeExternalSorter.java:399)
at
org.apache.spark.sql.execution.UnsafeExternalRowSorter.insertRow(UnsafeExternalRowSorter.java:94)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.sort_addToSorter$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:370)
at
org.apache.spark.sql.execution.RowIteratorFromScala.advanceNext(RowIterator.scala:83)
at
org.apache.spark.sql.execution.joins.SortMergeJoinScanner.advancedBufferedToRowWithNullFreeJoinKey(SortMergeJoinExec.scala:730)
at
org.apache.spark.sql.execution.joins.SortMergeJoinScanner.(SortMergeJoinExec.scala:605)
at
org.apache.spark.sql.execution.joins.SortMergeJoinExec$$anonfun$doExecute$1.apply(SortMergeJoinExec.scala:162)
at
org.apache.spark.sql.execution.joins.SortMergeJoinExec$$anonfun$doExecute$1.apply(SortMergeJoinExec.scala:100)
at
org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:89)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
at
org.apache.spark.rdd.CoalescedRDD$$anonfun$compute$1.apply(CoalescedRDD.scala:96)
at
org.apache.spark.rdd.CoalescedRDD$$anonfun$compute$1.apply(CoalescedRDD.scala:95)
at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:106)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:98)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:214)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:935)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:926)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:866)
at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:926)
at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:670)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:330)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:281)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
at org.apache.spark.scheduler.Task.run(Task.scala:86)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api sti
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15823297#comment-15823297
]
Lenni Kuff commented on SPARK-14560:
Hi,I will be out of the office from 12/19/16 - 1/20/17. The Hive, Sentry, Pig,
and RecordService teams have been transitioned to new leads who should be the
primary contacts moving forward:
*Sentry - Mat Crocker
*Hive - Sangeeta Doraiswamy
*RecordService - Alexander Bibighaus
*Pig - Ferenc Denes
*Everything Else - Eli CollinsAs usual, [1]Non-8x5 Escalations for each
component should be routed using the instructions on the wiki. I will be in
the NY area for most of my time away. If there is a critical item that comes up
you can reach me on my cell @ [2]415-840-4577.
Thanks,Lenni
[1] https://wiki.cloudera.com/pages/viewpage.action?pageId=24935790
[2] tel:415-840-4577
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15944077#comment-15944077
]
Darshan Mehta commented on SPARK-14560:
---
[~lovemylover] [~mhornbech] Were you able to figure out/fix the issue? We are
using Spark 2.1.0 and are facing similar bug, below is the stacktrace:
Caused by: java.lang.OutOfMemoryError: Unable to acquire 65536 bytes of memory,
got 0
at
org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:100)
at
org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.(UnsafeInMemorySorter.java:125)
at
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.(UnsafeExternalSorter.java:154)
at
org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:121)
at
org.apache.spark.sql.execution.UnsafeExternalRowSorter.(UnsafeExternalRowSorter.java:82)
at
org.apache.spark.sql.execution.SortExec.createSorter(SortExec.scala:87)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.init(Unknown
Source)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8.apply(WholeStageCodegenExec.scala:374)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8.apply(WholeStageCodegenExec.scala:371)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$26.apply(RDD.scala:843)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$26.apply(RDD.scala:843)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fa
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15944080#comment-15944080
]
Morten Hornbech commented on SPARK-14560:
-
No, not really. We were able to work around it by rewriting the job, but it was
never clear what made the difference.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15944090#comment-15944090
]
Darshan Mehta commented on SPARK-14560:
---
[~mhornbech] thanks for the prompt response. Before rewriting the job, did you
see any difference by playing around with
spark.shuffle.spill.numElementsForceSpillThreshold=N values?
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIR
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15944094#comment-15944094
]
Morten Hornbech commented on SPARK-14560:
-
No. We worried it might just trigger some other bad behaviour, and it wasn't a
production issue.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
-
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15948578#comment-15948578
]
Lianhui Wang commented on SPARK-14560:
--
[~darshankhamar123] [~mhornbech] Are you using ExternalAppendOnlyMap or
ExternalSorter? This issue is about them, not for UnsafeInMemorySorter.
You can fetch more information about memory using Debug level log.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15322857#comment-15322857
]
Peter Halliday commented on SPARK-14560:
Is this going to be in 1.6.2 too? Is there a timeline for when it might make
it in 1.6.2?
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To u
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15432406#comment-15432406
]
Sean Owen commented on SPARK-14560:
---
Even with the fix for SPARK-4452, I observed a problem like what's described
here: running out of memory in the shuffle read phase rather inexplicably,
which was worked around with numElementsForceSpillThreshold.
It turned out that enabling Java serialization caused the problem to go away
entirely. This was in Spark 1.6. No idea why, but leaving this as a note for
anyone who may find it, or if we later connect the dots elsewhere to an
underlying problem.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; t
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15432883#comment-15432883
]
Imran Rashid commented on SPARK-14560:
--
One minor clarification -- SPARK-4452 is not included in Spark 1.6, but Sean
was running a version with that fix backported. So if you see this problem in
Spark 1.6, (a) considering backporting SPARK-4452 and then try switching to
java serialization.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for fu
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15433200#comment-15433200
]
Davies Liu commented on SPARK-14560:
Even with SPARK-4452, we still can not say that we fixed the OOM problem
totally, because the memory used by Java object (used by RDD) can't be measured
and predicted exactly, in the case that they use memory than we thought, it
will still OOM. The java serializer may use less memory than Kyro, so it helped
in this case. Also, there are many memory are tracked in memory manager, for
example, the buffer used by shuffle reader, they could also go beyond the
capacity we preserved for all others, that could be also another cause for the
OOM in large scale job. The workaround could be decreasing the memoryFaction to
preserve more memory for all other stuff. Have you tried that?
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15433210#comment-15433210
]
Sean Owen commented on SPARK-14560:
---
I have a few somewhat-specific additional data points:
More memory didn't seem to help. A job that ran comfortably with tens of
gigabytes total with Java serialization would fail even with almost a terabyte
of memory available. The memory fraction was at the default of 0.75, or up to
0.9. I don't think we tried less, on the theory that the shuffle memory ought
to be tracked as part of the 'storage' memory?
But the same thing happened with the legacy memory manager.
Unhelpfully, the heap appeared full of byte[] and String.
The shuffle involved user classes that were reasonably complex: nested objects
involving case classes, third-party library classes, etc. None of them were
registered with Kryo. I tried registering most of them, on the theory that this
was causing some in-memory serialized representation to become huge. It didn't
seem to help, but I still wonder if there's a lead there. When Kryo doesn't
know about a class it serializes its class name first, but not the class names
of everything in the graph (right?) so it can only make so much difference.
Java serialization does the same.
For the record, it's just this Spark app that reproduces it:
https://github.com/sryza/aas/blob/master/ch08-geotime/src/main/scala/com/cloudera/datascience/geotime/RunGeoTime.scala
I have not tried on Spark 2, only 1.6 (CDH 5.8 flavor).
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Lianhui Wang
> Fix For: 2.0.0
>
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write sid
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15237226#comment-15237226
]
Imran Rashid commented on SPARK-14560:
--
I have a fix and have verified it works on real workloads -- I'm working on
adding some unit tests and will submit a PR soon.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Imran Rashid
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To u
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15239559#comment-15239559
]
Imran Rashid commented on SPARK-14560:
--
After some doing some more testing, I realized my "fix" wasn't really a fix at
all -- its not too hard to force the {{Spillable}} to spill, but the problem is
if you've already got an iterator, that keeps a reference to the in-memory
data. So it was really just cheating, by having the execution layer use more
memory than it was reporting (since it was reporting that it had freed memory
after the spill, when in fact it had not). Getting the inflight iterators to
switch from data in memory to the new spill is pretty gnarly, and I would need
to do a bunch of testing before I felt confident in it.
However I realized there are two solutions that are both much easier to
implement, safer, and can easily be put behind a conf to turn the behavior on
and off:
* always spill shuffle-read. Anytime there is a shuffle-to-stage stage and the
shuffle-read creates a {{Spillable}}, after its done inserting all records, it
can spill all of its data to disk. This will prevent starvation. The downside
is that you force spilling even if its not necessary. The benefit is that it
is really easy to do, and will prevent OOMs.
* reserve a fraction of execution memory for future operations. This is a
little trickier because (a) you've got to figure out if there are any future
execution operations in the same task and (b) come up with some sensible amount
of memory to reserve. Furthermore, though it can prevent an unnecessary spill
when there is enough memory, it actually may not be more efficient in all
scenarios. If there is a huge shuffle-read, which requires a lot of spills,
rather than keeping a little bit of it in memory at the very end, it probably
makes more sense to just create fewer spills in the first place by using all
available memory, and then spilling at the end to free memory for the next step.
true cooperative memory management for spillables would still be nice, but
given the complexity of that I think something simpler which will prevent OOM
is a better first step.
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Imran Rashid
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than wha
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15239573#comment-15239573
]
Imran Rashid commented on SPARK-14560:
--
Here's a reproduction. This could almost certainly be pared down to something
simpler, but figured this was worth sharing at this point. My initial
description of the issue was not entirely accurate -- there is something a bit
more subtle going on, it only happens with certain combinations of shuffle
operations, I'm still looking into that.
{code}
test("SPARK-14560") {
val myConf = conf.clone()
.set("spark.shuffle.memoryFraction", "0.01")
.set("spark.memory.useLegacyMode", "true")
.set("spark.testing.memory", "5") //~500MB
.set("spark.shuffle.sort.bypassMergeThreshold", "0")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //
for relocation, so we can use ShuffleExternalSorter
sc = new SparkContext("local", "test", myConf)
val N = 2e6.toInt
val p = new org.apache.spark.HashPartitioner(10)
val d = sc.parallelize(1 to N, 10).map { x => (x % 1) -> x.toLong }
// now we use an aggregator, but that still produces enough data that we
need to spill on the read-side
// (this one is ridiculous, we shouldn't aggregate at all, but its just an
easy
// way to trigger lots of memory use on the shuffle-read side)
val d2: RDD[(Int, Seq[Long])] = d.aggregateByKey(Seq[Long](), 5) (
{ case (list, next) => list :+ next },
{ case (listA, listB) => listA ++ listB }
)
val d3 = d2.repartitionAndSortWithinPartitions(p)
d3.count()
}
{code}
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Imran Rashid
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all
[jira] [Commented] (SPARK-14560) Cooperative Memory Management for Spillables
[
https://issues.apache.org/jira/browse/SPARK-14560?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15240047#comment-15240047
]
Apache Spark commented on SPARK-14560:
--
User 'squito' has created a pull request for this issue:
https://github.com/apache/spark/pull/12369
> Cooperative Memory Management for Spillables
>
>
> Key: SPARK-14560
> URL: https://issues.apache.org/jira/browse/SPARK-14560
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
>Affects Versions: 1.6.1
>Reporter: Imran Rashid
>Assignee: Imran Rashid
>
> SPARK-10432 introduced cooperative memory management for SQL operators that
> can spill; however, {{Spillable}} s used by the old RDD api still do not
> cooperate. This can lead to memory starvation, in particular on a
> shuffle-to-shuffle stage, eventually resulting in errors like:
> {noformat}
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Memory used in task 3081
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: Acquired by
> org.apache.spark.shuffle.sort.ShuffleExternalSorter@69ab0291: 32.0 KB
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317230346 bytes of memory
> were used by task 3081 but are not associated with specific consumers
> 16/03/28 08:59:54 INFO memory.TaskMemoryManager: 1317263114 bytes of memory
> are used for execution and 1710484 bytes of memory are used for storage
> 16/03/28 08:59:54 ERROR executor.Executor: Managed memory leak detected; size
> = 1317230346 bytes, TID = 3081
> 16/03/28 08:59:54 ERROR executor.Executor: Exception in task 533.0 in stage
> 3.0 (TID 3081)
> java.lang.OutOfMemoryError: Unable to acquire 75 bytes of memory, got 0
> at
> org.apache.spark.memory.MemoryConsumer.allocatePage(MemoryConsumer.java:120)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.acquireNewPageIfNecessary(ShuffleExternalSorter.java:346)
> at
> org.apache.spark.shuffle.sort.ShuffleExternalSorter.insertRecord(ShuffleExternalSorter.java:367)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.insertRecordIntoSorter(UnsafeShuffleWriter.java:237)
> at
> org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:164)
> 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:89)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:745)
> {noformat}
> This can happen anytime the shuffle read side requires more memory than what
> is available for the task. Since the shuffle-read side doubles its memory
> request each time, it can easily end up acquiring all of the available
> memory, even if it does not use it. Eg., say that after the final spill, the
> shuffle-read side requires 10 MB more memory, and there is 15 MB of memory
> available. But if it starts at 2 MB, it will double to 4, 8, and then
> request 16 MB of memory, and in fact get all available 15 MB. Since the 15
> MB of memory is sufficient, it will not spill, and will continue holding on
> to all available memory. But this leaves *no* memory available for the
> shuffle-write side. Since the shuffle-write side cannot request the
> shuffle-read side to free up memory, this leads to an OOM.
> The simple solution is to make {{Spillable}} implement {{MemoryConsumer}} as
> well, so RDDs can benefit from the cooperative memory management introduced
> by SPARK-10342.
> Note that an additional improvement would be for the shuffle-read side to
> simple release unused memory, without spilling, in case that would leave
> enough memory, and only spill if that was inadequate. However that can come
> as a later improvement.
> *Workaround*: You can set
> {{spark.shuffle.spill.numElementsForceSpillThreshold=N}} to force spilling to
> occur every {{N}} elements, thus preventing the shuffle-read side from ever
> grabbing all of the available memory. However, this requires careful tuning
> of {{N}} to specific workloads: too big, and you will still get an OOM; too
> small, and there will be so much spilling that performance will suffer
> drastically. Furthermore, this workaround uses an *undocumented*
> configuration with *no compatibility guarantees* for future versions of spark.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-
