{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/23083","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/23083/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/23083/comments","events_url":"https://api.github.com/repos/apache/spark/issues/23083/events","html_url":"https://github.com/apache/spark/pull/23083","id":382168300,"node_id":"MDExOlB1bGxSZXF1ZXN0MjMxOTM2OTIz","number":23083,"title":"[SPARK-26114][CORE] ExternalSorter's readingIterator field leak","user":{"login":"szhem","id":1523889,"node_id":"MDQ6VXNlcjE1MjM4ODk=","avatar_url":"https://avatars1.githubusercontent.com/u/1523889?v=4","gravatar_id":"","url":"https://api.github.com/users/szhem","html_url":"https://github.com/szhem","followers_url":"https://api.github.com/users/szhem/followers","following_url":"https://api.github.com/users/szhem/following{/other_user}","gists_url":"https://api.github.com/users/szhem/gists{/gist_id}","starred_url":"https://api.github.com/users/szhem/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/szhem/subscriptions","organizations_url":"https://api.github.com/users/szhem/orgs","repos_url":"https://api.github.com/users/szhem/repos","events_url":"https://api.github.com/users/szhem/events{/privacy}","received_events_url":"https://api.github.com/users/szhem/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":11,"created_at":"2018-11-19T11:19:13Z","updated_at":"2018-11-26T07:54:55Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/23083","html_url":"https://github.com/apache/spark/pull/23083","diff_url":"https://github.com/apache/spark/pull/23083.diff","patch_url":"https://github.com/apache/spark/pull/23083.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis pull request fixes [SPARK-26114](https://issues.apache.org/jira/browse/SPARK-26114) issue that occurs when trying to reduce the number of partitions by means of coalesce without shuffling after shuffle-based transformations. \r\n\r\nThe leak occurs because of not cleaning up `ExternalSorter`'s `readingIterator` field as it's done for its `map` and `buffer` fields.\r\nAdditionally there are changes to the `CompletionIterator` to prevent capturing its `sub`-iterator and holding it even after the completion iterator completes. It is necessary because in some cases, e.g. in case of standard scala's `flatMap` iterator (which is used is `CoalescedRDD`'s `compute` method) the next value of the main iterator is assigned to `flatMap`'s `cur` field only after it is available. \r\nFor DAGs where ShuffledRDD is a parent of CoalescedRDD it means that the data should be fetched from the map-side of the shuffle, but the process of fetching this data consumes quite a lot of memory in addition to the memory already consumed by the iterator held by `flatMap`'s `cur` field (until it is reassigned).\r\n\r\nFor the following data\r\n```scala\r\nimport org.apache.hadoop.io._ \r\nimport org.apache.hadoop.io.compress._ \r\nimport org.apache.commons.lang._ \r\nimport org.apache.spark._ \r\n\r\n// generate 100M records of sample data \r\nsc.makeRDD(1 to 1000, 1000) \r\n  .flatMap(item => (1 to 100000) \r\n    .map(i => new Text(RandomStringUtils.randomAlphanumeric(3).toLowerCase) -> new Text(RandomStringUtils.randomAlphanumeric(1024)))) \r\n  .saveAsSequenceFile(\"/tmp/random-strings\", Some(classOf[GzipCodec])) \r\n```\r\n\r\nand the following job\r\n```scala\r\nimport org.apache.hadoop.io._\r\nimport org.apache.spark._\r\nimport org.apache.spark.storage._\r\n\r\nval rdd = sc.sequenceFile(\"/tmp/random-strings\", classOf[Text], classOf[Text])\r\nrdd \r\n  .map(item => item._1.toString -> item._2.toString) \r\n  .repartitionAndSortWithinPartitions(new HashPartitioner(1000)) \r\n  .coalesce(10,false) \r\n  .count \r\n```\r\n\r\n... executed like the following\r\n```bash\r\nspark-shell \\ \r\n  --num-executors=5 \\ \r\n  --executor-cores=2 \\ \r\n  --master=yarn \\\r\n  --deploy-mode=client \\ \r\n  --conf spark.executor.memoryOverhead=512 \\\r\n  --conf spark.executor.memory=1g \\ \r\n  --conf spark.dynamicAllocation.enabled=false \\\r\n  --conf spark.executor.extraJavaOptions='-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp -Dio.netty.noUnsafe=true'\r\n```\r\n\r\n... executors are always failing with OutOfMemoryErrors.\r\n\r\nThe main issue is multiple leaks of ExternalSorter references.\r\nFor example, in case of 2 tasks per executor it is expected to be 2 simultaneous instances of ExternalSorter per executor but heap dump generated on OutOfMemoryError shows that there are more ones.\r\n\r\n![run1-noparams-dominator-tree-externalsorter](https://user-images.githubusercontent.com/1523889/48703665-782ce580-ec05-11e8-95a9-d6c94e8285ab.png)\r\n\r\n\r\nP.S. This PR does not cover cases with CoGroupedRDDs which use ExternalAppendOnlyMap internally, which itself can lead to OutOfMemoryErrors in many places. \r\n\r\n## How was this patch tested?\r\n\r\n- Existing unit tests\r\n- New unit tests\r\n- Job executions on the live environment\r\n\r\nHere is the screenshot before applying this patch\r\n![run3-noparams-failure-ui-5x2-repartition-and-sort](https://user-images.githubusercontent.com/1523889/48700395-f769eb80-ebfc-11e8-831b-e94c757d416c.png)\r\n\r\nHere is the screenshot after applying this patch\r\n![run3-noparams-success-ui-5x2-repartition-and-sort](https://user-images.githubusercontent.com/1523889/48700610-7a8b4180-ebfd-11e8-9761-baaf38a58e66.png)\r\nAnd in case of reducing the number of executors even more the job is still stable\r\n![run3-noparams-success-ui-2x2-repartition-and-sort](https://user-images.githubusercontent.com/1523889/48700619-82e37c80-ebfd-11e8-98ed-a38e1f1f1fd9.png)\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/441548082","html_url":"https://github.com/apache/spark/pull/23083#issuecomment-441548082","issue_url":"https://api.github.com/repos/apache/spark/issues/23083","id":441548082,"node_id":"MDEyOklzc3VlQ29tbWVudDQ0MTU0ODA4Mg==","user":{"login":"AmplabJenkins","id":4031291,"node_id":"MDQ6VXNlcjQwMzEyOTE=","avatar_url":"https://avatars3.githubusercontent.com/u/4031291?v=4","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":"2018-11-26T07:54:55Z","updated_at":"2018-11-26T07:54:55Z","author_association":"NONE","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}