{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/58253255","id":58253255,"diff_hunk":"@@ -477,12 +605,16 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr\n private[spark] object MapOutputTracker extends Logging {\n \n   val ENDPOINT_NAME = \"MapOutputTracker\"\n+  private val DIRECT = 0\n+  private val BROADCAST = 1\n \n   // Serialize an array of map output locations into an efficient byte format so that we can send\n   // it to reduce tasks. We do this by compressing the serialized bytes using GZIP. They will\n   // generally be pretty compressible because many map outputs will be on the same hostname.\n-  def serializeMapStatuses(statuses: Array[MapStatus]): Array[Byte] = {\n+  def serializeMapStatuses(statuses: Array[MapStatus], broadcastManager: BroadcastManager,\n+      isLocal: Boolean, minBroadcastSize: Int): (Array[Byte], Broadcast[Array[Byte]]) = {","path":"core/src/main/scala/org/apache/spark/MapOutputTracker.scala","position":285,"original_position":285,"commit_id":"3c1def02b80ebfed55904e504609aa02de6559ce","original_commit_id":"3c1def02b80ebfed55904e504609aa02de6559ce","user":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"body":"Both values are used in getSerializedMapOutputStatuses","created_at":"2016-04-01T19:09:40Z","updated_at":"2016-04-01T19:09:40Z","html_url":"https://github.com/apache/spark/pull/12113#discussion_r58253255","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/12113","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/58253255"},"html":{"href":"https://github.com/apache/spark/pull/12113#discussion_r58253255"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/12113"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/12113","id":64988721,"html_url":"https://github.com/apache/spark/pull/12113","diff_url":"https://github.com/apache/spark/pull/12113.diff","patch_url":"https://github.com/apache/spark/pull/12113.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/12113","number":12113,"state":"open","locked":false,"title":"[SPARK-1239] Improve fetching of map output statuses","user":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"body":"The main issue we are trying to solve is the memory bloat of the Driver when tasks request the map output statuses.  This means with a large number of tasks you either need a huge amount of memory on Driver or you have to repartition to smaller number.  This makes it really difficult to run over say 50000 tasks.\r\n\r\nThe main issues that cause the memory bloat are:\r\n1) no flow control on sending the map output status responses.  We serialize the map status output  and then hand off to netty to send.  netty is sending asynchronously and it can't send them fast enough to keep up with incoming requests so we end up with lots of copies of the serialized map output statuses sitting there and this causes huge bloat when you have 10's of thousands of tasks and map output status is in the 10's of MB.\r\n2) When initial reduce tasks are started up, they all request the map output statuses from the Driver. These requests are handled by multiple threads in parallel so even though we check to see if we have a cached version, initially when we don't have a cached version yet, many of initial requests can all end up serializing the exact same map output statuses.\r\n\r\nThis patch does a couple of things:\r\n- When the map output status size is over a threshold (default 512K) then it uses broadcast to send the map statuses.  This means we no longer serialize a large map output status and thus we don't have issues with memory bloat.  the messages sizes are now in the 300-400 byte range and the map status output are broadcast. If its under the threadshold it sends it as before, the message contains the DIRECT indicator now.\r\n- synchronize the incoming requests to allow one thread to cache the serialized output and broadcast the map output status  that can then be used by everyone else.  This ensures we don't create multiple broadcast variables when we don't need to.  To ensure this happens I added a second thread pool which the Dispatcher hands the requests to so that those threads can block without blocking the main dispatcher threads (which would cause things like heartbeats and such not to come through)\r\n\r\nNote that some of design and code was contributed by @mridulm\r\n\r\n## How was this patch tested?\r\n\r\nUnit tests and a lot of manually testing. \r\nRan with akka and netty rpc. Ran with both dynamic allocation on and off.\r\n\r\none of the large jobs I used to test this was a join of 15TB of data.  it had 200,000 map tasks, and  20,000 reduce tasks. Executors ranged from 200 to 2000.  This job ran successfully with 5GB of memory on the driver with these changes. Without these changes I was using 20GB and only had 500 reduce tasks.  The job has 50mb of serialized map output statuses and took roughly the same amount of time for the executors to get the map output statuses as before.\r\n\r\nRan a variety of other jobs, from large wordcounts to small ones not using broadcasts. \r\n","created_at":"2016-04-01T18:51:17Z","updated_at":"2016-04-01T19:09:40Z","closed_at":null,"merged_at":null,"merge_commit_sha":"60e39351bc43780bde3982921193722f5f406aa5","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/12113/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/12113/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/12113/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/3c1def02b80ebfed55904e504609aa02de6559ce","head":{"label":"tgravescs:SPARK-1239","ref":"SPARK-1239","sha":"3c1def02b80ebfed55904e504609aa02de6559ce","user":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"repo":{"id":17249522,"name":"spark","full_name":"tgravescs/spark","owner":{"login":"tgravescs","id":4563792,"avatar_url":"https://avatars.githubusercontent.com/u/4563792?v=3","gravatar_id":"","url":"https://api.github.com/users/tgravescs","html_url":"https://github.com/tgravescs","followers_url":"https://api.github.com/users/tgravescs/followers","following_url":"https://api.github.com/users/tgravescs/following{/other_user}","gists_url":"https://api.github.com/users/tgravescs/gists{/gist_id}","starred_url":"https://api.github.com/users/tgravescs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tgravescs/subscriptions","organizations_url":"https://api.github.com/users/tgravescs/orgs","repos_url":"https://api.github.com/users/tgravescs/repos","events_url":"https://api.github.com/users/tgravescs/events{/privacy}","received_events_url":"https://api.github.com/users/tgravescs/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/tgravescs/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/tgravescs/spark","forks_url":"https://api.github.com/repos/tgravescs/spark/forks","keys_url":"https://api.github.com/repos/tgravescs/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tgravescs/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tgravescs/spark/teams","hooks_url":"https://api.github.com/repos/tgravescs/spark/hooks","issue_events_url":"https://api.github.com/repos/tgravescs/spark/issues/events{/number}","events_url":"https://api.github.com/repos/tgravescs/spark/events","assignees_url":"https://api.github.com/repos/tgravescs/spark/assignees{/user}","branches_url":"https://api.github.com/repos/tgravescs/spark/branches{/branch}","tags_url":"https://api.github.com/repos/tgravescs/spark/tags","blobs_url":"https://api.github.com/repos/tgravescs/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tgravescs/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tgravescs/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/tgravescs/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tgravescs/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/tgravescs/spark/languages","stargazers_url":"https://api.github.com/repos/tgravescs/spark/stargazers","contributors_url":"https://api.github.com/repos/tgravescs/spark/contributors","subscribers_url":"https://api.github.com/repos/tgravescs/spark/subscribers","subscription_url":"https://api.github.com/repos/tgravescs/spark/subscription","commits_url":"https://api.github.com/repos/tgravescs/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/tgravescs/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/tgravescs/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/tgravescs/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/tgravescs/spark/contents/{+path}","compare_url":"https://api.github.com/repos/tgravescs/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tgravescs/spark/merges","archive_url":"https://api.github.com/repos/tgravescs/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tgravescs/spark/downloads","issues_url":"https://api.github.com/repos/tgravescs/spark/issues{/number}","pulls_url":"https://api.github.com/repos/tgravescs/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/tgravescs/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/tgravescs/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tgravescs/spark/labels{/name}","releases_url":"https://api.github.com/repos/tgravescs/spark/releases{/id}","deployments_url":"https://api.github.com/repos/tgravescs/spark/deployments","created_at":"2014-02-27T13:26:24Z","updated_at":"2015-05-08T13:44:25Z","pushed_at":"2016-04-01T18:50:10Z","git_url":"git://github.com/tgravescs/spark.git","ssh_url":"git@github.com:tgravescs/spark.git","clone_url":"https://github.com/tgravescs/spark.git","svn_url":"https://github.com/tgravescs/spark","homepage":null,"size":143151,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"8ba2b7f28fee39c4839e5ea125bd25f5091a3a1e","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2016-04-01T18:35:39Z","pushed_at":"2016-04-01T18:51:18Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":447952,"stargazers_count":8005,"watchers_count":8005,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":7385,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":430,"forks":7385,"open_issues":430,"watchers":8005,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/12113"},"html":{"href":"https://github.com/apache/spark/pull/12113"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/12113"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/12113/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/12113/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/12113/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/3c1def02b80ebfed55904e504609aa02de6559ce"}}}}