{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/212209237","pull_request_review_id":148789601,"id":212209237,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDIxMjIwOTIzNw==","diff_hunk":"@@ -1502,6 +1502,53 @@ private[spark] class DAGScheduler(\n             failedStages += failedStage\n             failedStages += mapStage\n             if (noResubmitEnqueued) {\n+              // If the map stage is INDETERMINATE, which means the map tasks may return\n+              // different result when re-try, we need to re-try all the tasks of the failed\n+              // stage and its succeeding stages, because the input data will be changed after the\n+              // map tasks are re-tried.\n+              // Note that, if map stage is UNORDERED, we are fine. The shuffle partitioner is\n+              // guaranteed to be idempotent, so the input data of the reducers will not change even\n+              // if the map tasks are re-tried.\n+              if (mapStage.rdd.computingRandomLevel == RDD.RandomLevel.INDETERMINATE) {\n+                def rollBackStage(stage: Stage): Unit = stage match {\n+                  case mapStage: ShuffleMapStage =>\n+                    val numMissingPartitions = mapStage.findMissingPartitions().length\n+                    if (numMissingPartitions < mapStage.numTasks) {\n+                      markStageAsFinished(\n+                        mapStage,\n+                        Some(\"preceding shuffle map stage with random output gets retried.\"),\n+                        willRetry = true)\n+                      mapOutputTracker.unregisterAllMapOutput(mapStage.shuffleDep.shuffleId)\n+                      failedStages += mapStage\n+                    }\n+\n+                  case resultStage =>\n+                    val numMissingPartitions = resultStage.findMissingPartitions().length\n+                    if (numMissingPartitions < resultStage.numTasks) {\n+                      // TODO: support to rollback result tasks.\n+                      val errorMessage = \"A shuffle map stage with random output was failed and \" +\n+                        s\"retried. However, Spark cannot rollback the result stage $resultStage \" +\n+                        \"to re-process the input data, and has to fail this job. Please \" +\n+                        \"eliminate the randomness by checkpointing the RDD before \" +\n+                        \"repartition/zip and try again.\"\n+                      abortStage(failedStage, errorMessage, None)\n+                    }\n+                }\n+\n+                def rollbackSucceedingStages(stageChain: List[Stage]): Unit = {\n+                  if (stageChain.head.id == failedStage.id) {\n+                    stageChain.foreach { stage =>\n+                      if (!failedStages.contains(stage)) rollBackStage(stage)\n+                    }\n+                  } else {\n+                    stageChain.head.parents.foreach(s => rollbackSucceedingStages(s :: stageChain))\n+                  }\n+                }","path":"core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala","position":45,"original_position":45,"commit_id":"93f37fa585462b9ee2fb9e179eab736fbc416d3e","original_commit_id":"93f37fa585462b9ee2fb9e179eab736fbc416d3e","user":{"login":"cloud-fan","id":3182036,"node_id":"MDQ6VXNlcjMxODIwMzY=","avatar_url":"https://avatars3.githubusercontent.com/u/3182036?v=4","gravatar_id":"","url":"https://api.github.com/users/cloud-fan","html_url":"https://github.com/cloud-fan","followers_url":"https://api.github.com/users/cloud-fan/followers","following_url":"https://api.github.com/users/cloud-fan/following{/other_user}","gists_url":"https://api.github.com/users/cloud-fan/gists{/gist_id}","starred_url":"https://api.github.com/users/cloud-fan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloud-fan/subscriptions","organizations_url":"https://api.github.com/users/cloud-fan/orgs","repos_url":"https://api.github.com/users/cloud-fan/repos","events_url":"https://api.github.com/users/cloud-fan/events{/privacy}","received_events_url":"https://api.github.com/users/cloud-fan/received_events","type":"User","site_admin":false},"body":"On the other hand, this branch will not be hitten very offten. We only hit it when a FetchFailure happens and the map stage has random output.","created_at":"2018-08-23T07:36:37Z","updated_at":"2018-08-23T07:36:37Z","html_url":"https://github.com/apache/spark/pull/22112#discussion_r212209237","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/22112","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/212209237"},"html":{"href":"https://github.com/apache/spark/pull/22112#discussion_r212209237"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/22112"}},"in_reply_to_id":212197939},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/22112","id":208665959,"node_id":"MDExOlB1bGxSZXF1ZXN0MjA4NjY1OTU5","html_url":"https://github.com/apache/spark/pull/22112","diff_url":"https://github.com/apache/spark/pull/22112.diff","patch_url":"https://github.com/apache/spark/pull/22112.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/22112","number":22112,"state":"open","locked":false,"title":"[SPARK-23243][Core] Fix RDD.repartition() data correctness issue","user":{"login":"cloud-fan","id":3182036,"node_id":"MDQ6VXNlcjMxODIwMzY=","avatar_url":"https://avatars3.githubusercontent.com/u/3182036?v=4","gravatar_id":"","url":"https://api.github.com/users/cloud-fan","html_url":"https://github.com/cloud-fan","followers_url":"https://api.github.com/users/cloud-fan/followers","following_url":"https://api.github.com/users/cloud-fan/following{/other_user}","gists_url":"https://api.github.com/users/cloud-fan/gists{/gist_id}","starred_url":"https://api.github.com/users/cloud-fan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloud-fan/subscriptions","organizations_url":"https://api.github.com/users/cloud-fan/orgs","repos_url":"https://api.github.com/users/cloud-fan/repos","events_url":"https://api.github.com/users/cloud-fan/events{/privacy}","received_events_url":"https://api.github.com/users/cloud-fan/received_events","type":"User","site_admin":false},"body":"## What changes were proposed in this pull request?\r\n\r\nAn alternative fix for https://github.com/apache/spark/pull/21698\r\n\r\nThere are 3 kinds of RDD computing functions, according to their behavior when rerun:\r\n1. Idempotent. Always return the same result with same order when rerun.\r\n2. Random order. Returns same data set in random order when rerun.\r\n3. Completely random. Returns different result when rerun.\r\n\r\nNormally Spark doesn't need to care about it. Spark runs stages one by one, when a task is failed, just rerun it. Although the rerun task may return a different result, users will not be surprised.\r\n\r\nHowever, Spark may rerun a finished stage when seeing fetch failures. When this happens, Spark needs to rerun all the tasks of all the succeeding stages if the computing function is completely random, because the input of these stages has been changed.\r\n\r\nIf the computing function is idempotent, we only need to rerun the failed tasks of the succeeding stages, because the input doesn't change.\r\n\r\nIf the computing function outputs same data set in a random order, it's same as idempotent, because shuffle partitioner is always idempotent(round-robin partitioner is not a shuffle partitioner that extends `org.apache.spark.Partitioner`), so the reducers will still get the same input data set.\r\n\r\nThis PR fixed the failure handling for `repartition` as well as `zip`, to avoid correctness issues.\r\n\r\nFor `repartition`, it applies a stateful map function to generate a round-robin id, which is order sensitive and makes the RDD's output completely random. When the stage contains `repartition` reruns, we must also rerun all the tasks of all the succeeding stages.\r\n\r\nFor `zip`, it's map function is `(iter1, iter2) => iter1.zip(iter2)`. It will return completely random output if either `iter1` or `iter2` changes its output order.\r\n\r\n**future improvement:**\r\n1. Currently we can't rollback and rerun a result stage, and just fail. We should fix it later.\r\n2. We should provide public API to allow users to tag the random level of the RDD's computing function.\r\n\r\n\r\n## How is this pull request tested?\r\na new test case","created_at":"2018-08-15T20:02:51Z","updated_at":"2018-08-23T07:36:37Z","closed_at":null,"merged_at":null,"merge_commit_sha":"914ae7870a587a2a1436e187da849a5c4f04dff9","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/22112/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/22112/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/22112/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/93f37fa585462b9ee2fb9e179eab736fbc416d3e","head":{"label":"cloud-fan:repartition","ref":"repartition","sha":"93f37fa585462b9ee2fb9e179eab736fbc416d3e","user":{"login":"cloud-fan","id":3182036,"node_id":"MDQ6VXNlcjMxODIwMzY=","avatar_url":"https://avatars3.githubusercontent.com/u/3182036?v=4","gravatar_id":"","url":"https://api.github.com/users/cloud-fan","html_url":"https://github.com/cloud-fan","followers_url":"https://api.github.com/users/cloud-fan/followers","following_url":"https://api.github.com/users/cloud-fan/following{/other_user}","gists_url":"https://api.github.com/users/cloud-fan/gists{/gist_id}","starred_url":"https://api.github.com/users/cloud-fan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloud-fan/subscriptions","organizations_url":"https://api.github.com/users/cloud-fan/orgs","repos_url":"https://api.github.com/users/cloud-fan/repos","events_url":"https://api.github.com/users/cloud-fan/events{/privacy}","received_events_url":"https://api.github.com/users/cloud-fan/received_events","type":"User","site_admin":false},"repo":{"id":19689699,"node_id":"MDEwOlJlcG9zaXRvcnkxOTY4OTY5OQ==","name":"spark","full_name":"cloud-fan/spark","owner":{"login":"cloud-fan","id":3182036,"node_id":"MDQ6VXNlcjMxODIwMzY=","avatar_url":"https://avatars3.githubusercontent.com/u/3182036?v=4","gravatar_id":"","url":"https://api.github.com/users/cloud-fan","html_url":"https://github.com/cloud-fan","followers_url":"https://api.github.com/users/cloud-fan/followers","following_url":"https://api.github.com/users/cloud-fan/following{/other_user}","gists_url":"https://api.github.com/users/cloud-fan/gists{/gist_id}","starred_url":"https://api.github.com/users/cloud-fan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloud-fan/subscriptions","organizations_url":"https://api.github.com/users/cloud-fan/orgs","repos_url":"https://api.github.com/users/cloud-fan/repos","events_url":"https://api.github.com/users/cloud-fan/events{/privacy}","received_events_url":"https://api.github.com/users/cloud-fan/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/cloud-fan/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/cloud-fan/spark","forks_url":"https://api.github.com/repos/cloud-fan/spark/forks","keys_url":"https://api.github.com/repos/cloud-fan/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/cloud-fan/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/cloud-fan/spark/teams","hooks_url":"https://api.github.com/repos/cloud-fan/spark/hooks","issue_events_url":"https://api.github.com/repos/cloud-fan/spark/issues/events{/number}","events_url":"https://api.github.com/repos/cloud-fan/spark/events","assignees_url":"https://api.github.com/repos/cloud-fan/spark/assignees{/user}","branches_url":"https://api.github.com/repos/cloud-fan/spark/branches{/branch}","tags_url":"https://api.github.com/repos/cloud-fan/spark/tags","blobs_url":"https://api.github.com/repos/cloud-fan/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/cloud-fan/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/cloud-fan/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/cloud-fan/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/cloud-fan/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/cloud-fan/spark/languages","stargazers_url":"https://api.github.com/repos/cloud-fan/spark/stargazers","contributors_url":"https://api.github.com/repos/cloud-fan/spark/contributors","subscribers_url":"https://api.github.com/repos/cloud-fan/spark/subscribers","subscription_url":"https://api.github.com/repos/cloud-fan/spark/subscription","commits_url":"https://api.github.com/repos/cloud-fan/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/cloud-fan/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/cloud-fan/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/cloud-fan/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/cloud-fan/spark/contents/{+path}","compare_url":"https://api.github.com/repos/cloud-fan/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/cloud-fan/spark/merges","archive_url":"https://api.github.com/repos/cloud-fan/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/cloud-fan/spark/downloads","issues_url":"https://api.github.com/repos/cloud-fan/spark/issues{/number}","pulls_url":"https://api.github.com/repos/cloud-fan/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/cloud-fan/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/cloud-fan/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/cloud-fan/spark/labels{/name}","releases_url":"https://api.github.com/repos/cloud-fan/spark/releases{/id}","deployments_url":"https://api.github.com/repos/cloud-fan/spark/deployments","created_at":"2014-05-12T07:32:26Z","updated_at":"2017-06-22T06:57:26Z","pushed_at":"2018-08-23T00:44:42Z","git_url":"git://github.com/cloud-fan/spark.git","ssh_url":"git@github.com:cloud-fan/spark.git","clone_url":"https://github.com/cloud-fan/spark.git","svn_url":"https://github.com/cloud-fan/spark","homepage":null,"size":272574,"stargazers_count":1,"watchers_count":1,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":1,"open_issues":0,"watchers":1,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"0295ad40def41b9a8ccefaaa1a7658899fb632a4","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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,"node_id":"MDEwOlJlcG9zaXRvcnkxNzE2NTY1OA==","name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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":"2018-08-23T07:12:42Z","pushed_at":"2018-08-23T07:09:08Z","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":280242,"stargazers_count":18485,"watchers_count":18485,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":16720,"mirror_url":null,"archived":false,"open_issues_count":475,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":16720,"open_issues":475,"watchers":18485,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/22112"},"html":{"href":"https://github.com/apache/spark/pull/22112"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/22112"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/22112/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/22112/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/22112/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/93f37fa585462b9ee2fb9e179eab736fbc416d3e"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"cloud-fan"}}