Tue Nov 8 20:09:31 PST 2016

Author: ericl

core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala

@@ -84,7 +84,8 @@ abstract class FileCommitProtocol {
    * are left to the commit protocol implementation to decide.
    *
    * Important: it is the caller's responsibility to add uniquely identifying content to "ext"
-   * if a task is going to write out multiple files to the same dir.
+   * if a task is going to write out multiple files to the same dir. The file commit protocol only
+   * guarantees that files written by different tasks will not conflict.
    */
   def newTaskTempFile(taskContext: TaskAttemptContext, dir: Option[String], ext: String): String
 
@@ -93,7 +94,8 @@ abstract class FileCommitProtocol {
    * Depending on the implementation, there may be weaker guarantees around adding files this way.
    *
    * Important: it is the caller's responsibility to add uniquely identifying content to "ext"
-   * if a task is going to write out multiple files at all (even to different directories).
+   * if a task is going to write out multiple files to the same dir. The file commit protocol only
+   * guarantees that files written by different tasks will not conflict.
    */
   def newTaskTempFileAbsPath(
       taskContext: TaskAttemptContext, absoluteDir: String, ext: String): String

core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.internal.io
 
-import java.util.Date
+import java.util.{Date, UUID}
 
 import scala.collection.mutable
 
@@ -82,7 +82,12 @@ class HadoopMapReduceCommitProtocol(jobId: String, path: String)
       taskContext: TaskAttemptContext, absoluteDir: String, ext: String): String = {
     val filename = getFilename(taskContext, ext)
     val absOutputPath = new Path(absoluteDir, filename).toString
-    val tmpOutputPath = new Path(absPathStagingDir, filename).toString
+
+    // Include a UUID here to prevent file collisions for one task writing to different dirs.
+    // In principle we could include hash(absoluteDir) instead but this is simpler.
+    val tmpOutputPath = new Path(
+      absPathStagingDir, UUID.randomUUID().toString() + "-" + filename).toString
+
     addedAbsPathFiles(tmpOutputPath) = absOutputPath
     tmpOutputPath
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala

@@ -321,10 +321,7 @@ object FileFormatWriter extends Logging {
           None
       }
       val path = if (customPath.isDefined) {
-        // We need to include a uuid here since the commit protocol does not guarantee that
-        // temp files requested by the same task for absolute placement do not collide.
-        committer.newTaskTempFileAbsPath(
-          taskAttemptContext, customPath.get, UUID.randomUUID().toString + ext)
+        committer.newTaskTempFileAbsPath(taskAttemptContext, customPath.get, ext)
       } else {
         committer.newTaskTempFile(taskAttemptContext, partDir, ext)
       }