[
https://issues.apache.org/jira/browse/SPARK-19659?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15952401#comment-15952401
]
Imran Rashid commented on SPARK-19659:
--------------------------------------
[~cloud_fan] The memory is allocated by netty, but its still possible to track
it. Netty already lets you *poll* its memory usage, which gives a crude way to
track the memory. That is being looked at in SPARK-9103. It already has an
old pr which I think was close, but is now out-of-date; [~jsoltren] is looking
at bringing it up to date.
A bigger change would be to have closer interaction between netty and spark's
memory management. Netty actually already has an extension point where spark
could do this -- its {{ByteBufAllocator}}. Spark already configures the
ByteBufAllocator, though it just chooses one of the existing implementations:
https://github.com/apache/spark/blob/master/common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java#L95
https://github.com/apache/spark/blob/master/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java#L107
We could swap in our own ByteBufAllocator -- it could even just extend netty's
{{PooledByteBufAllocator}} that we're using now, just track allocations as they
happen; or even, check with the MemoryManager on each allocation whether there
is memory available, and if not, don't do the fetch yet. Obviously those are
bigger changes, but I think it should be possible conceptually. I don't know
of any jiras to track that yet.
> Fetch big blocks to disk when shuffle-read
> ------------------------------------------
>
> Key: SPARK-19659
> URL: https://issues.apache.org/jira/browse/SPARK-19659
> Project: Spark
> Issue Type: Improvement
> Components: Shuffle
> Affects Versions: 2.1.0
> Reporter: jin xing
> Attachments: SPARK-19659-design-v1.pdf, SPARK-19659-design-v2.pdf
>
>
> Currently the whole block is fetched into memory(offheap by default) when
> shuffle-read. A block is defined by (shuffleId, mapId, reduceId). Thus it can
> be large when skew situations. If OOM happens during shuffle read, job will
> be killed and users will be notified to "Consider boosting
> spark.yarn.executor.memoryOverhead". Adjusting parameter and allocating more
> memory can resolve the OOM. However the approach is not perfectly suitable
> for production environment, especially for data warehouse.
> Using Spark SQL as data engine in warehouse, users hope to have a unified
> parameter(e.g. memory) but less resource wasted(resource is allocated but not
> used),
> It's not always easy to predict skew situations, when happen, it make sense
> to fetch remote blocks to disk for shuffle-read, rather than
> kill the job because of OOM. This approach is mentioned during the discussion
> in SPARK-3019, by [~sandyr] and [~mridulm80]
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
