{"push_id":571144622,"size":1,"distinct_size":1,"ref":"refs/heads/master","head":"1cb37700753437045b15c457b983532cd5a27fa5","before":"7e24249af1e2f896328ef0402fa47db78cb6f9ec","commits":[{"sha":"1cb37700753437045b15c457b983532cd5a27fa5","author":{"email":"2e9bf58ec2f93170404b3a28538fb27400f699db@palantir.com","name":"mcheah"},"message":"[SPARK-4879] Use driver to coordinate Hadoop output committing for speculative tasks\n\nPreviously, SparkHadoopWriter always committed its tasks without question. The problem is that when speculation is enabled sometimes this can result in multiple tasks committing their output to the same file. 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.\n\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 SPARK-4879.\n\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.\n\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.\n\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.\n\nCurrently, the OutputCommitCoordinator is only used when `spark.speculation` is true.  It can be disabled by setting `spark.hadoop.outputCommitCoordination.enabled=false` in SparkConf.\n\nThis patch is an updated version of #4155 (by mccheah), which in turn was an updated version of this PR.\n\nCloses #4155.\n\nAuthor: mcheah <mcheah@palantir.com>\nAuthor: Josh Rosen <joshrosen@databricks.com>\n\nCloses #4066 from JoshRosen/SPARK-4879-sparkhadoopwriter-fix and squashes the following commits:\n\n658116b [Josh Rosen] Merge remote-tracking branch 'origin/master' into SPARK-4879-sparkhadoopwriter-fix\ned783b2 [Josh Rosen] Address Andrew’s feedback.\ne7be65a [Josh Rosen] Merge remote-tracking branch 'origin/master' into SPARK-4879-sparkhadoopwriter-fix\n14861ea [Josh Rosen] splitID -> partitionID in a few places\ned8b554 [Josh Rosen] Merge remote-tracking branch 'origin/master' into SPARK-4879-sparkhadoopwriter-fix\n48d5c1c [Josh Rosen] Roll back copiesRunning change in TaskSetManager\n3969f5f [Josh Rosen] Re-enable guarding of commit coordination with spark.speculation setting.\nede7590 [Josh Rosen] Add test to ensure that a job that denies all commits cannot complete successfully.\n97da5fe [Josh Rosen] Use actor only for RPC; call methods directly in DAGScheduler.\nf582574 [Josh Rosen] Some cleanup in OutputCommitCoordinatorSuite\na7c0e29 [Josh Rosen] Create fake TaskInfo using dummy fields instead of Mockito.\n997b41b [Josh Rosen] Roll back unnecessary DAGSchedulerSingleThreadedProcessLoop refactoring:\n459310a [Josh Rosen] Roll back TaskSetManager changes that broke other tests.\ndd00b7c [Josh Rosen] Move CommitDeniedException to executors package; remove `@DeveloperAPI` annotation.\nc79df98 [Josh Rosen] Some misc. code style + doc changes:\nf7d69c5 [Josh Rosen] Merge remote-tracking branch 'origin/master' into SPARK-4879-sparkhadoopwriter-fix\n92e6dc9 [Josh Rosen] Bug fix: use task ID instead of StageID to index into authorizedCommitters.\nb344bad [Josh Rosen] (Temporarily) re-enable “always coordinate” for testing purposes.\n0aec91e [Josh Rosen] Only coordinate when speculation is enabled; add configuration option to bypass new coordination.\n594e41a [mcheah] Fixing a scalastyle error\n60a47f4 [mcheah] Writing proper unit test for OutputCommitCoordinator and fixing bugs.\nd63f63f [mcheah] Fixing compiler error\n9fe6495 [mcheah] Fixing scalastyle\n1df2a91 [mcheah] Throwing exception if SparkHadoopWriter commit denied\nd431144 [mcheah] Using more concurrency to process OutputCommitCoordinator requests.\nc334255 [mcheah] Properly handling messages that could be sent after actor shutdown.\n8d5a091 [mcheah] Was mistakenly serializing the accumulator in test suite.\n9c6a4fa [mcheah] More OutputCommitCoordinator cleanup on stop()\n78eb1b5 [mcheah] Better OutputCommitCoordinatorActor stopping; simpler canCommit\n83de900 [mcheah] Making the OutputCommitCoordinatorMessage serializable\nabc7db4 [mcheah] TaskInfo can't be null in DAGSchedulerSuite\nf135a8e [mcheah] Moving the output commit coordinator from class into method.\n1c2b219 [mcheah] Renaming oudated names for test function classes\n66a71cd [mcheah] Removing whitespace modifications\n6b543ba [mcheah] Removing redundant accumulator in unit test\nc9decc6 [mcheah] Scalastyle fixes\nbc80770 [mcheah] Unit tests for OutputCommitCoordinator\n6e6f748 [mcheah] [SPARK-4879] Use the Spark driver to authorize Hadoop commits.","distinct":true,"url":"https://api.github.com/repos/apache/spark/commits/1cb37700753437045b15c457b983532cd5a27fa5"}]}