{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/4155","labels_url":"https://api.github.com/repos/apache/spark/issues/4155/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/4155/comments","events_url":"https://api.github.com/repos/apache/spark/issues/4155/events","html_url":"https://github.com/apache/spark/pull/4155","id":55105184,"number":4155,"title":"[SPARK-4879] Use the Spark driver to authorize Hadoop commits.","user":{"login":"mccheah","id":8975385,"avatar_url":"https://avatars.githubusercontent.com/u/8975385?v=3","gravatar_id":"","url":"https://api.github.com/users/mccheah","html_url":"https://github.com/mccheah","followers_url":"https://api.github.com/users/mccheah/followers","following_url":"https://api.github.com/users/mccheah/following{/other_user}","gists_url":"https://api.github.com/users/mccheah/gists{/gist_id}","starred_url":"https://api.github.com/users/mccheah/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mccheah/subscriptions","organizations_url":"https://api.github.com/users/mccheah/orgs","repos_url":"https://api.github.com/users/mccheah/repos","events_url":"https://api.github.com/users/mccheah/events{/privacy}","received_events_url":"https://api.github.com/users/mccheah/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":88,"created_at":"2015-01-22T02:01:55Z","updated_at":"2015-02-10T22:23:11Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/4155","html_url":"https://github.com/apache/spark/pull/4155","diff_url":"https://github.com/apache/spark/pull/4155.diff","patch_url":"https://github.com/apache/spark/pull/4155.patch"},"body":"This is a version of https://github.com/apache/spark/pull/4066/ which is up to date with master and has unit tests.\r\n\r\nPreviously, SparkHadoopWriter always committed its tasks without question. The problem is that when speculation is turned on, sometimes this can result in multiple tasks committing their output to the same\r\npartition. Even though an HDFS-writing task may be re-launched due to speculation, the original task is not killed and may eventually commit as well.\r\n\r\nThis can cause strange race conditions where multiple tasks that commit interfere with each other, with the result being that some partition files are actually lost entirely. For more context on these kinds of scenarios, see the aforementioned JIRA ticket.\r\n\r\nIn Hadoop MapReduce jobs, the application master is a central coordinator that authorizes whether or not any given task can commit. Before a task commits its output, it queries the application master as to whether or not such a commit is safe, and the application master does bookkeeping as tasks are requesting commits. Duplicate tasks that would write to files that were already written to from other tasks are prohibited from committing.\r\n\r\nThis patch emulates that functionality - the crucial missing component was a central arbitrator, which is now a module called the OutputCommitCoordinator. The coordinator lives on the driver and the executors can obtain a reference to this actor and request its permission to commit. As tasks commit and are reported as completed successfully or unsuccessfully by the DAGScheduler, the commit coordinator is informed of the task completion events as well to update its internal state.\r\n\r\nFuture work includes more rigorous unit testing and extra optimizations should this patch cause a performance regression. It is unclear what the overall cost of communicating back to the driver on every hadoop-committing task will be. It's also important for those hitting this issue to backport this onto previous version of Spark because the bug has serious consequences, that is, data is lost.\r\n\r\ncc @MingyuKim @JoshRosen  @ash211 "},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/73797137","html_url":"https://github.com/apache/spark/pull/4155#issuecomment-73797137","issue_url":"https://api.github.com/repos/apache/spark/issues/4155","id":73797137,"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},"created_at":"2015-02-10T22:23:11Z","updated_at":"2015-02-10T22:23:11Z","body":"([Cross-posted from JIRA](https://issues.apache.org/jira/browse/SPARK-4879?focusedCommentId=14315077&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14315077))\r\n\r\nThis issue is really hard to reproduce, but I managed to trigger the original bug as part of the testing for my patch. Here's what I ran:\r\n\r\n```bash\r\n~/spark-1.3.0-SNAPSHOT-bin-1.0.4/bin/spark-shell --conf spark.speculation.multiplier=1 --conf spark.speculation.quantile=0.01 --conf spark.speculation=true --conf  spark.hadoop.outputCommitCoordination.enabled=false\r\n```\r\n\r\n```scala\r\nval numTasks = 100\r\nval numTrials = 100\r\nval outputPath = \"/output-committer-bug-\"\r\nval sleepDuration = 1000\r\n\r\nfor (trial <- 0 to (numTrials - 1)) {\r\n  val outputLocation = outputPath + trial\r\n  sc.parallelize(1 to numTasks, numTasks).mapPartitionsWithContext { case (ctx, iter) =>\r\n    if (ctx.partitionId % 5 == 0) {\r\n      if (ctx.attemptNumber == 0) {  // If this is the first attempt, run slow\r\n       Thread.sleep(sleepDuration)\r\n      }\r\n    }\r\n    iter\r\n  }.map(identity).saveAsTextFile(outputLocation)\r\n  Thread.sleep(sleepDuration * 2)\r\n  println(\"TESTING OUTPUT OF TRIAL \" + trial)\r\n  val savedData = sc.textFile(outputLocation).map(_.toInt).collect()\r\n  if (savedData.toSet != (1 to numTasks).toSet) {\r\n    println(\"MISSING: \" + ((1 to numTasks).toSet -- savedData.toSet))\r\n    assert(false)\r\n  }\r\n  println(\"-\" * 80)\r\n}\r\n```\r\n\r\nIt took 22 runs until I actually observed missing output partitions (several of the earlier runs threw spurious exceptions and didn't have missing outputs):\r\n\r\n```\r\n[...]\r\n15/02/10 22:17:21 INFO scheduler.DAGScheduler: Job 66 finished: saveAsTextFile at <console>:39, took 2.479592 s\r\n15/02/10 22:17:21 WARN scheduler.TaskSetManager: Lost task 75.0 in stage 66.0 (TID 6861, ip-172-31-1-124.us-west-2.compute.internal): java.io.IOException: Failed to save output of task: attempt_201502102217_0066_m_000075_6861\r\n        at org.apache.hadoop.mapred.FileOutputCommitter.moveTaskOutputs(FileOutputCommitter.java:160)\r\n        at org.apache.hadoop.mapred.FileOutputCommitter.moveTaskOutputs(FileOutputCommitter.java:172)\r\n        at org.apache.hadoop.mapred.FileOutputCommitter.commitTask(FileOutputCommitter.java:132)\r\n        at org.apache.spark.SparkHadoopWriter.performCommit$1(SparkHadoopWriter.scala:113)\r\n        at org.apache.spark.SparkHadoopWriter.commit(SparkHadoopWriter.scala:150)\r\n        at org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:1082)\r\n        at org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:1059)\r\n        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\r\n        at org.apache.spark.scheduler.Task.run(Task.scala:64)\r\n        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:197)\r\n        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\r\n        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\r\n        at java.lang.Thread.run(Thread.java:745)\r\n\r\n15/02/10 22:17:21 WARN scheduler.TaskSetManager: Lost task 80.0 in stage 66.0 (TID 6866, ip-172-31-11-151.us-west-2.compute.internal): java.io.IOException: The temporary job-output directory hdfs://ec2-54-213-142-80.us-west-2.compute.amazonaws.com:9000/output-committer-bug-22/_temporary doesn't exist!\r\n        at org.apache.hadoop.mapred.FileOutputCommitter.getWorkPath(FileOutputCommitter.java:250)\r\n        at org.apache.hadoop.mapred.FileOutputFormat.getTaskOutputPath(FileOutputFormat.java:244)\r\n        at org.apache.hadoop.mapred.TextOutputFormat.getRecordWriter(TextOutputFormat.java:116)\r\n        at org.apache.spark.SparkHadoopWriter.open(SparkHadoopWriter.scala:91)\r\n        at org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:1068)\r\n        at org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:1059)\r\n        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\r\n        at org.apache.spark.scheduler.Task.run(Task.scala:64)\r\n        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:197)\r\n        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\r\n        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\r\n        at java.lang.Thread.run(Thread.java:745)\r\n\r\n15/02/10 22:17:21 INFO scheduler.TaskSetManager: Lost task 85.0 in stage 66.0 (TID 6871) on executor ip-172-31-1-124.us-west-2.compute.internal: java.io.IOException (The temporary job-output directory hdfs://ec2-54-213-142-80.us-west-2.compute.amazonaws.com:9000/output-committer-bug-22/_temporary doesn't exist!) [duplicate 1]\r\n15/02/10 22:17:21 INFO scheduler.TaskSetManager: Lost task 90.0 in stage 66.0 (TID 6876) on executor ip-172-31-1-124.us-west-2.compute.internal: java.io.IOException (The temporary job-output directory hdfs://ec2-54-213-142-80.us-west-2.compute.amazonaws.com:9000/output-committer-bug-22/_temporary doesn't exist!) [duplicate 2]\r\n15/02/10 22:17:21 INFO scheduler.TaskSetManager: Lost task 95.0 in stage 66.0 (TID 6881) on executor ip-172-31-11-151.us-west-2.compute.internal: java.io.IOException (The temporary job-output directory hdfs://ec2-54-213-142-80.us-west-2.compute.amazonaws.com:9000/output-committer-bug-22/_temporary doesn't exist!) [duplicate 3]\r\n15/02/10 22:17:21 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 66.0, whose tasks have all completed, from pool\r\nTESTING OUTPUT OF TRIAL 22\r\n[...]\r\n15/02/10 22:17:23 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 67.0, whose tasks have all completed, from pool\r\n15/02/10 22:17:23 INFO scheduler.DAGScheduler: Stage 67 (collect at <console>:42) finished in 0.158 s\r\n15/02/10 22:17:23 INFO scheduler.DAGScheduler: Job 67 finished: collect at <console>:42, took 0.165580 s\r\nMISSING: Set(76)\r\njava.lang.AssertionError: assertion failed\r\n[...]\r\n```\r\n\r\nAnd I confirmed that it's missing in HDFS:\r\n\r\n```\r\n~/ephemeral-hdfs/bin/hadoop fs -ls /output-committer-bug-22 | grep part | wc -l\r\nWarning: $HADOOP_HOME is deprecated.\r\n\r\n99\r\n```\r\n\r\nTo test my patch, I'm going to remove the flag that disables it and run this for a huge number of trials to ensure that we don't hit the missing output bug.\r\n"}}