spark git commit: [SPARK-18219] Move commit protocol API (internal) from sql/core to core module
Repository: spark
Updated Branches:
refs/heads/master 96cc1b567 -> 937af592e
[SPARK-18219] Move commit protocol API (internal) from sql/core to core module
## What changes were proposed in this pull request?
This patch moves the new commit protocol API from sql/core to core module, so
we can use it in the future in the RDD API.
As part of this patch, I also moved the speficiation of the random uuid for the
write path out of the commit protocol, and instead pass in a job id.
## How was this patch tested?
N/A
Author: Reynold Xin
Closes #15731 from rxin/SPARK-18219.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/937af592
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/937af592
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/937af592
Branch: refs/heads/master
Commit: 937af592e65f4dd878aafcabf8fe2cfe7fa3d9b3
Parents: 96cc1b56
Author: Reynold Xin
Authored: Thu Nov 3 02:42:48 2016 -0700
Committer: Reynold Xin
Committed: Thu Nov 3 02:42:48 2016 -0700
--
.../spark/internal/io/FileCommitProtocol.scala | 126 +
.../io/HadoopMapReduceCommitProtocol.scala | 111
.../datasources/FileCommitProtocol.scala| 257 ---
.../datasources/FileFormatWriter.scala | 3 +-
.../InsertIntoHadoopFsRelationCommand.scala | 6 +-
.../SQLHadoopMapReduceCommitProtocol.scala | 72 ++
.../execution/streaming/FileStreamSink.scala| 9 +-
.../streaming/ManifestFileCommitProtocol.scala | 6 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 4 +-
9 files changed, 327 insertions(+), 267 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/937af592/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
--
diff --git
a/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
b/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
new file mode 100644
index 000..fb80205
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.internal.io
+
+import org.apache.hadoop.mapreduce._
+
+import org.apache.spark.util.Utils
+
+
+/**
+ * An interface to define how a single Spark job commits its outputs. Two
notes:
+ *
+ * 1. Implementations must be serializable, as the committer instance
instantiated on the driver
+ *will be used for tasks on executors.
+ * 2. Implementations should have a constructor with either 2 or 3 arguments:
+ *(jobId: String, path: String) or (jobId: String, path: String, isAppend:
Boolean).
+ * 3. A committer should not be reused across multiple Spark jobs.
+ *
+ * The proper call sequence is:
+ *
+ * 1. Driver calls setupJob.
+ * 2. As part of each task's execution, executor calls setupTask and then
commitTask
+ *(or abortTask if task failed).
+ * 3. When all necessary tasks completed successfully, the driver calls
commitJob. If the job
+ *failed to execute (e.g. too many failed tasks), the job should call
abortJob.
+ */
+abstract class FileCommitProtocol {
+ import FileCommitProtocol._
+
+ /**
+ * Setups up a job. Must be called on the driver before any other methods
can be invoked.
+ */
+ def setupJob(jobContext: JobContext): Unit
+
+ /**
+ * Commits a job after the writes succeed. Must be called on the driver.
+ */
+ def commitJob(jobContext: JobContext, taskCommits: Seq[TaskCommitMessage]):
Unit
+
+ /**
+ * Aborts a job after the writes fail. Must be called on the driver.
+ *
+ * Calling this function is a best-effort attempt, because it is possible
that the driver
+ * just crashes (or killed) before it can call abort.
+ */
+ def abortJob(jobContext: JobContext): Unit
+
+ /**
+ * Sets up a task within a job.
+ * Must be called before any other task related methods can be invoked.
+ */
+ def setupTask(taskContext: TaskAttemptCon
spark git commit: [SPARK-18219] Move commit protocol API (internal) from sql/core to core module
Repository: spark
Updated Branches:
refs/heads/branch-2.1 c4c5328f2 -> bc7f05f5f
[SPARK-18219] Move commit protocol API (internal) from sql/core to core module
## What changes were proposed in this pull request?
This patch moves the new commit protocol API from sql/core to core module, so
we can use it in the future in the RDD API.
As part of this patch, I also moved the speficiation of the random uuid for the
write path out of the commit protocol, and instead pass in a job id.
## How was this patch tested?
N/A
Author: Reynold Xin
Closes #15731 from rxin/SPARK-18219.
(cherry picked from commit 937af592e65f4dd878aafcabf8fe2cfe7fa3d9b3)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/bc7f05f5
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/bc7f05f5
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/bc7f05f5
Branch: refs/heads/branch-2.1
Commit: bc7f05f5f03653c623190b8178bcbe981a41c2f3
Parents: c4c5328
Author: Reynold Xin
Authored: Thu Nov 3 02:42:48 2016 -0700
Committer: Reynold Xin
Committed: Thu Nov 3 02:43:03 2016 -0700
--
.../spark/internal/io/FileCommitProtocol.scala | 126 +
.../io/HadoopMapReduceCommitProtocol.scala | 111
.../datasources/FileCommitProtocol.scala| 257 ---
.../datasources/FileFormatWriter.scala | 3 +-
.../InsertIntoHadoopFsRelationCommand.scala | 6 +-
.../SQLHadoopMapReduceCommitProtocol.scala | 72 ++
.../execution/streaming/FileStreamSink.scala| 9 +-
.../streaming/ManifestFileCommitProtocol.scala | 6 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 4 +-
9 files changed, 327 insertions(+), 267 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/bc7f05f5/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
--
diff --git
a/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
b/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
new file mode 100644
index 000..fb80205
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.internal.io
+
+import org.apache.hadoop.mapreduce._
+
+import org.apache.spark.util.Utils
+
+
+/**
+ * An interface to define how a single Spark job commits its outputs. Two
notes:
+ *
+ * 1. Implementations must be serializable, as the committer instance
instantiated on the driver
+ *will be used for tasks on executors.
+ * 2. Implementations should have a constructor with either 2 or 3 arguments:
+ *(jobId: String, path: String) or (jobId: String, path: String, isAppend:
Boolean).
+ * 3. A committer should not be reused across multiple Spark jobs.
+ *
+ * The proper call sequence is:
+ *
+ * 1. Driver calls setupJob.
+ * 2. As part of each task's execution, executor calls setupTask and then
commitTask
+ *(or abortTask if task failed).
+ * 3. When all necessary tasks completed successfully, the driver calls
commitJob. If the job
+ *failed to execute (e.g. too many failed tasks), the job should call
abortJob.
+ */
+abstract class FileCommitProtocol {
+ import FileCommitProtocol._
+
+ /**
+ * Setups up a job. Must be called on the driver before any other methods
can be invoked.
+ */
+ def setupJob(jobContext: JobContext): Unit
+
+ /**
+ * Commits a job after the writes succeed. Must be called on the driver.
+ */
+ def commitJob(jobContext: JobContext, taskCommits: Seq[TaskCommitMessage]):
Unit
+
+ /**
+ * Aborts a job after the writes fail. Must be called on the driver.
+ *
+ * Calling this function is a best-effort attempt, because it is possible
that the driver
+ * just crashes (or killed) before it can call abort.
+ */
+ def abortJob(jobContext: JobContext): Unit
+
+ /**
+ * Sets up a task within a job.
+ * Must be called
