{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/11805","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/11805/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/11805/comments","events_url":"https://api.github.com/repos/apache/spark/issues/11805/events","html_url":"https://github.com/apache/spark/pull/11805","id":141740744,"number":11805,"title":"[SPARK-13992][WIP] Add support for off-heap caching","user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":12,"created_at":"2016-03-18T00:13:48Z","updated_at":"2016-03-27T04:12:07Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/11805","html_url":"https://github.com/apache/spark/pull/11805","diff_url":"https://github.com/apache/spark/pull/11805.diff","patch_url":"https://github.com/apache/spark/pull/11805.patch"},"body":"This patch adds support for caching blocks in the executor processes using direct / off-heap memory.\r\n\r\nThis is WIP because it's rebased on top of #11942, the patch to add off-heap storage memory bookkeeping support to MemoryManager. \r\n\r\n## User-facing changes\r\n\r\n**Updated semantics of `OFF_HEAP` storage level**: In Spark 1.x, the `OFF_HEAP` storage level indicated that an RDD should be cached in Tachyon. Spark 2.x removed the external block store API that Tachyon caching was based on (see #10752 / SPARK-12667), so `OFF_HEAP` became an alias for `MEMORY_ONLY_SER`. As of this patch, `OFF_HEAP` means \"serialized and cached in off-heap memory or on disk\". Via the `StorageLevel` constructor, `useOffHeap` can be set if `serialized == true` and can be used to construct custom storage levels which support replication.\r\n\r\n**Storage UI reporting**: the storage UI will now report whether in-memory blocks are stored on- or off-heap.\r\n\r\n**Only supported by UnifiedMemoryManager**: for simplicity, this feature is only supported when the default UnifiedMemoryManager is used; applications which use the legacy memory manager (`spark.memory.useLegacyMode=true`) are not currently able to allocate off-heap storage memory, so using off-heap caching will fail with an error when legacy memory management is enabled. Given that we plan to eventually remove the legacy memory manager, this is not a significant restriction.\r\n\r\n**Memory management policies:** the policies for dividing available memory between execution and storage are the same for both on- and off-heap memory. For off-heap memory, the total amount of memory available for use by Spark is controlled by `spark.memory.offHeap.size`, which is an absolute size. Off-heap storage memory obeys `spark.memory.storageFraction` in order to control the amount of unevictable storage memory. For example, if `spark.memory.offHeap.size` is 1 gigabyte and Spark uses the default `storageFraction` of 0.5, then up to 500 megabytes of off-heap cached blocks will be protected from eviction due to execution memory pressure. If necessary, we can split `spark.memory.storageFraction` into separate on- and off-heap configurations, but this doesn't seem necessary now and can be done later without any breaking changes.\r\n\r\n**Use of off-heap memory does not imply use of off-heap execution (or vice-versa)**: for now, the settings controlling the use of off-heap execution memory (`spark.memory.offHeap.enabled`) and off-heap caching are completely independent, so Spark SQL can be configured to use off-heap memory for execution while continuing to cache blocks on-heap. If desired, we can change this in a followup patch so that `spark.memory.offHeap.enabled` affect the default storage level for cached SQL tables.\r\n\r\n## Internal changes\r\n\r\n- Rename `ByteArrayChunkOutputStream` to `ChunkedByteBufferOutputStream`\r\n  - It now returns a `ChunkedByteBuffer` instead of an array of byte arrays.\r\n  - Its constructor now accept an `allocator` function which is called to allocate `ByteBuffer`s. This allows us to control whether it allocates regular ByteBuffers or off-heap DirectByteBuffers.\r\n  - Because block serialization is now performed during the unroll process, a `ChunkedByteBufferOutputStream` which is configured with a `DirectByteBuffer` allocator will use off-heap memory for both unroll and storage memory.\r\n- The `MemoryStore`'s MemoryEntries now tracks whether blocks are stored on- or off-heap.\r\n  - `evictBlocksToFreeSpace()` now accepts a `MemoryMode` parameter so that we don't try to evict off-heap blocks in response to on-heap memory pressure (or vice-versa).\r\n- Make sure that off-heap buffers are properly de-allocated during MemoryStore eviction.\r\n\r\n## Open questions\r\n\r\n- [ ] The JVM limits the total size of allocated direct byte buffers using the `-XX:MaxDirectMemorySize` flag and the default tends to be fairly low (< 512 megabytes in some JVMs). There are two options to work around this limitation:\r\n\r\n  1. Modify the `launcher` code to increase this limit to a large value. This is what I did for now, but it has potential downsides in case third-party libraries react poorly to this value being too high (e.g. allocate too much memory and OOM the container) or in case third-party launcher code doesn't set the JVM options or overrides them.\r\n  2. Build our own DirectByteBuffer allocator which ignores this memory limit. I think that this is achievable if we're willing to rely on some JDK internals for constructing DirectByteBuffers from pointers (we could allocate the memory ourselves using `sun.misc.Unsafe`)."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/201984448","html_url":"https://github.com/apache/spark/pull/11805#issuecomment-201984448","issue_url":"https://api.github.com/repos/apache/spark/issues/11805","id":201984448,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2016-03-27T04:12:06Z","updated_at":"2016-03-27T04:12:06Z","body":"Merged build finished. Test PASSed."}}