[SPARK-18024][SQL] Introduce an internal commit protocol API #15696
Conversation
|
cc @ericl |
|
Test build #67818 has finished for PR 15696 at commit
|
|
Test build #67832 has finished for PR 15696 at commit
|
|
Test build #67835 has finished for PR 15696 at commit
|
|
|
||
| override def getDefaultWorkFile(context: TaskAttemptContext, extension: String): Path = { | ||
| new Path(stagingDir, fileNamePrefix) | ||
| new Path(path) |
There was a problem hiding this comment.
Nope -- no more extension coming from Hadoop.
There was a problem hiding this comment.
(This is now specified explicitly in OutputWriterFactory.getFileExtension)
|
Test build #67838 has finished for PR 15696 at commit
|
|
Test build #67840 has finished for PR 15696 at commit
|
rxin
changed the title
|
|
||
|
|
||
| /** | ||
| * An interface to define how a Spark job commits its outputs. Implementations must be serializable. |
There was a problem hiding this comment.
add: since the same committer instance setup on the driver will be used for tasks.
| * The "dir" parameter specifies 2, and "ext" parameter specifies both 4 and 5, and the rest | ||
| * are left to the commit protocol implementation to decide. | ||
| */ | ||
| def addTaskTempFile(taskContext: TaskAttemptContext, dir: Option[String], ext: String): String |
|
|
||
| /** | ||
| * Notifies the commit protocol to add a new file, and gets back the full path that should be | ||
| * used. Must be called on the executors when running tasks. |
There was a problem hiding this comment.
add: Note that the returned temp file may have an arbitrary path. The commit protocol only promises that the file will be at the location specified by the arguments after job commit.
| /** | ||
| * Aborts a task after the writes have failed. Must be called on the executors when running tasks. | ||
| */ | ||
| def abortTask(taskContext: TaskAttemptContext): Unit |
| * | ||
| * Unlike Hadoop's OutputCommitter, this implementation is serializable. | ||
| */ | ||
| class MapReduceFileCommitterProtocol(path: String, isAppend: Boolean) |
There was a problem hiding this comment.
Should we call this HadoopCommitProtocolWrapper or something to be more clear?
| final def filePrefix(split: Int, uuid: String, bucketId: Option[Int]): String = { | ||
| val bucketString = bucketId.map(BucketingUtils.bucketIdToString).getOrElse("") | ||
| f"part-r-$split%05d-$uuid$bucketString" | ||
| f"part-$split%05d-$uuid$bucketString" |
|
Test build #67841 has finished for PR 15696 at commit
|
|
Test build #67852 has finished for PR 15696 at commit
|
|
Closing this in favor of #15707 |
4 participants
What changes were proposed in this pull request?
This patch introduces an internal commit protocol API that is used by the batch data source to do write commits. It currently has only one implementation that uses Hadoop MapReduce's OutputCommitter API. In the future, this commit API can be used to unify streaming and batch commits.
How was this patch tested?
Should be covered by existing write tests.