You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/08/10 14:37:51 UTC
[GitHub] [spark] steveloughran opened a new pull request, #37468: [WIP][SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
steveloughran opened a new pull request, #37468:
URL: https://github.com/apache/spark/pull/37468
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Uses the StreamCapabilities probe in MAPREDUCE-7403 to identify when a
PathOutputCommitter is compatible with dynamic partition overwrite.
This patch has unit tests but not integration tests; really needs
to test the SQL commands through the manifest committer into gcs/abfs,
or at least local fs. That would be possible once hadoop 3.3.5 is out...
Uses the StreamCapabilities probe in MAPREDUCE-7403 to identify when a
PathOutputCommitter is compatible with dynamic partition overwrite.
### Why are the changes needed?
Hadoop 3.3.5 adds a new committer in mapreduce-core which works fast and correctly on azure and gcs. (it would also work on hdfs, but its optimised for the cloud stores).
The stores and the committer do meet the requirements of Spark SQL Dynamic Partition Overwrite, so it is OK to for spark to work through it.
Spark does not know this; MAPREDUCE-7403 adds a way for any PathOutputCommitter to declare that they are compatible; the IntermediateManifestCommitter will do so.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
This patch has unit tests but not integration tests; really needs
to test the SQL commands through the manifest committer into gcs/abfs,
or at least local fs. That would be possible once hadoop 3.3.5 is out.
I do plan to do an end to end test; this pr is a WiP until then
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] attilapiros commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
attilapiros commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964890599
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -76,6 +78,16 @@ class CommitterBindingSuite extends SparkFunSuite {
assert(inner.jobCommitted, s"$inner job not committed")
parquet.abortJob(tContext, JobStatus.State.RUNNING)
assert(inner.jobAborted, s"$inner job not aborted")
+
+ val binding = new BindingPathOutputCommitter(path, tContext)
+ if (binding.isInstanceOf[StreamCapabilities]) {
Review Comment:
I see.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] attilapiros commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
attilapiros commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964330609
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -125,21 +132,57 @@ class CommitterBindingSuite extends SparkFunSuite {
val instance = FileCommitProtocol.instantiate(
pathCommitProtocolClassname,
jobId, "file:///tmp", false)
-
val protocol = instance.asInstanceOf[PathOutputCommitProtocol]
assert("file:///tmp" === protocol.destination)
}
- test("reject dynamic partitioning") {
- val cause = intercept[InvocationTargetException] {
- FileCommitProtocol.instantiate(
- pathCommitProtocolClassname,
- jobId, "file:///tmp", true)
- }.getCause
- if (cause == null || !cause.isInstanceOf[IOException]
- || !cause.getMessage.contains(PathOutputCommitProtocol.UNSUPPORTED)) {
- throw cause
+ test("reject dynamic partitioning if not supported") {
+ val path = new Path("http://example/data")
+ val job = newJob(path)
+ val conf = job.getConfiguration
+ conf.set(MRJobConfig.TASK_ATTEMPT_ID, taskAttempt0)
+ conf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 1)
+ StubPathOutputCommitterBinding.bind(conf, "http")
+ val tContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+ val committer = FileCommitProtocol.instantiate(
+ pathCommitProtocolClassname,
+ jobId, path.toUri.toString, true)
+ val ioe = intercept[IOException] {
+ committer.setupJob(tContext)
+ }
+ if (!ioe.getMessage.contains(PathOutputCommitProtocol.UNSUPPORTED)) {
+ throw ioe
}
+ intercept[UnsupportedOperationException] {
+ committer.newTaskTempFileAbsPath(
+ tContext,
+ "/tmp",
+ FileNameSpec("lotus", ".123"))
+ }
+ }
+
+ test("permit dynamic partitioning if the committer says it works") {
Review Comment:
Could you please add a new test where the exact same things are checked just using `FileOutputCommitterFactory` at the binding (by extracting the test body into separate method(s) and reusing most of the parts)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964967014
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -125,21 +132,57 @@ class CommitterBindingSuite extends SparkFunSuite {
val instance = FileCommitProtocol.instantiate(
pathCommitProtocolClassname,
jobId, "file:///tmp", false)
-
val protocol = instance.asInstanceOf[PathOutputCommitProtocol]
assert("file:///tmp" === protocol.destination)
}
- test("reject dynamic partitioning") {
- val cause = intercept[InvocationTargetException] {
- FileCommitProtocol.instantiate(
- pathCommitProtocolClassname,
- jobId, "file:///tmp", true)
- }.getCause
- if (cause == null || !cause.isInstanceOf[IOException]
- || !cause.getMessage.contains(PathOutputCommitProtocol.UNSUPPORTED)) {
- throw cause
+ test("reject dynamic partitioning if not supported") {
+ val path = new Path("http://example/data")
+ val job = newJob(path)
+ val conf = job.getConfiguration
+ conf.set(MRJobConfig.TASK_ATTEMPT_ID, taskAttempt0)
+ conf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 1)
+ StubPathOutputCommitterBinding.bind(conf, "http")
+ val tContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+ val committer = FileCommitProtocol.instantiate(
+ pathCommitProtocolClassname,
+ jobId, path.toUri.toString, true)
+ val ioe = intercept[IOException] {
+ committer.setupJob(tContext)
+ }
+ if (!ioe.getMessage.contains(PathOutputCommitProtocol.UNSUPPORTED)) {
+ throw ioe
}
+ intercept[UnsupportedOperationException] {
+ committer.newTaskTempFileAbsPath(
+ tContext,
+ "/tmp",
+ FileNameSpec("lotus", ".123"))
+ }
+ }
+
+ test("permit dynamic partitioning if the committer says it works") {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1242332651
Thank you, @steveloughran and @attilapiros .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1241751151
@dongjoon-hyun I'm off on vacation next week; we will fork off the branch the week after.
things i'd like in *if anyone has the time*
1. upgraded shaded parquet
2. the shaded avro PR
3. get that arm64 docker image working for a release there
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1228570834
the hadoop side of this change is now merged in.
@attilapiros do you have any time to review this again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r965456236
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -161,7 +204,16 @@ object PathOutputCommitProtocol {
val REJECT_FILE_OUTPUT_DEFVAL = false
/** Error string for tests. */
- private[cloud] val UNSUPPORTED: String = "PathOutputCommitProtocol does not support" +
+ private[cloud] val UNSUPPORTED: String = "PathOutputCommitter does not support" +
" dynamicPartitionOverwrite"
+ /**
+ * Stream Capabilities probe for spark dynamic partitioning compatibility.
Review Comment:
indentation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964636364
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -114,10 +112,33 @@ class PathOutputCommitProtocol(
// failures. Warn
logTrace(s"Committer $committer may not be tolerant of task commit failures")
}
+ } else {
+ // if required other committers need to be checked for dynamic partition
+ // compatibility through a StreamCapabilities probe.
+ if (dynamicPartitionOverwrite) {
+ if (supportsDynamicPartitions) {
+ logDebug(
+ s"Committer $committer has declared compatibility with dynamic partition overwrite")
+ } else {
+ throw new IOException(PathOutputCommitProtocol.UNSUPPORTED + ": " + committer)
+ }
+ }
}
committer
}
+
+ /**
+ * Does the instantiated committer support dynamic partitions?
+ * @return true if the committer declares itself compatible.
+ */
+ private def supportsDynamicPartitions = {
+ committer.isInstanceOf[FileOutputCommitter] ||
Review Comment:
up until now, if you used file output committer through the factory, it would reject the use of dynamic partitions. now it lets the caller without needing to implement the path capability.
this does ensure that you can use that committer through the factory mechanism even on hadoop versions <= 3.3.4. I didn't add the path capabilities probe to the original FileOutputCommitter because I'm scared of going near that file: old, complex, critical code
+ will update the user facing details
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1241563463
BTW, when is the ETA for Apache Hadoop 3.3.5, @steveloughran ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1242329502
Merged to master for Apache Spark 3.4.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r949986859
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -140,6 +162,28 @@ class PathOutputCommitProtocol(
file.toString
}
+ /**
+ * Reject any requests for an absolute path file on a committer which
+ * is not compatible with it.
+ *
+ * @param taskContext task context
+ * @param absoluteDir final directory
+ * @param spec output filename
+ * @return a path string
+ * @throws UnsupportedOperationException if incompatible
+ */
+ override def newTaskTempFileAbsPath(
+ taskContext: TaskAttemptContext,
+ absoluteDir: String,
+ spec: FileNameSpec): String = {
+
+ if (supportsDynamicPartitions) {
+ super.newTaskTempFileAbsPath(taskContext, absoluteDir, spec)
+ } else {
+ throw new UnsupportedOperationException(s"Absolute output locations not supported" +
+ s" by committer $committer")
+ }
+ }
Review Comment:
adding a safety defence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1448575588
@attilapiros FYI, there is a new PR for this area.
- https://github.com/apache/spark/pull/40221
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] attilapiros commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
attilapiros commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964124161
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -114,10 +112,33 @@ class PathOutputCommitProtocol(
// failures. Warn
logTrace(s"Committer $committer may not be tolerant of task commit failures")
}
+ } else {
+ // if required other committers need to be checked for dynamic partition
+ // compatibility through a StreamCapabilities probe.
+ if (dynamicPartitionOverwrite) {
+ if (supportsDynamicPartitions) {
+ logDebug(
+ s"Committer $committer has declared compatibility with dynamic partition overwrite")
+ } else {
+ throw new IOException(PathOutputCommitProtocol.UNSUPPORTED + ": " + committer)
+ }
+ }
}
committer
}
+
+ /**
+ * Does the instantiated committer support dynamic partitions?
+ * @return true if the committer declares itself compatible.
+ */
+ private def supportsDynamicPartitions = {
+ committer.isInstanceOf[FileOutputCommitter] ||
Review Comment:
Do we change the old behaviour because of the `committer.isInstanceOf[FileOutputCommitter]` part of the condition? As before this PR we were simply throwing an exception.
If this is a change then the `Does this PR introduce any user-facing change?` section of the PR description must be updated too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1448602874
This is reverted from `branch-3.4` only while being kept in `master` branch for Apache Spark 3.5.
- https://github.com/apache/spark/commit/57aa3d1ca6e17e4c6b934d74176ea22ca56d60f7
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1217015149
this should interest @sunchao and @dongjoon-hyun -know that this doesn't add support to the s3a committers; s3 itself doesn't do the right thing (rename()). Does for abfs and gcs through the manifest committer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r965454379
##########
docs/cloud-integration.md:
##########
@@ -266,26 +271,123 @@ It has been tested with the most common formats supported by Spark.
mydataframe.write.format("parquet").save("s3a://bucket/destination")
```
-More details on these committers can be found in the latest Hadoop documentation.
+More details on these committers can be found in
+[the latest Hadoop documentation](https://hadoop.apache.org/docs/current/)
+with S3A committer detail covered in
+[Committing work to S3 with the S3A Committers](https://hadoop.apache.org/docs/current/hadoop-aws/tools/hadoop-aws/committers.html).
Note: depending upon the committer used, in-progress statistics may be
under-reported with Hadoop versions before 3.3.1.
+### Amazon EMR: the EMRFS S3-optimized committer
+
+Amazon EMR has its own S3-aware committers for parquet data.
+For instructions on use, see
+[the EMRFS S3-optimized committer](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-s3-optimized-committer.html)
+
+For implementation and performanc details, see
+["Improve Apache Spark write performance on Apache Parquet formats with the EMRFS S3-optimized committer"](https://aws.amazon.com/blogs/big-data/improve-apache-spark-write-performance-on-apache-parquet-formats-with-the-emrfs-s3-optimized-committer/
+
+
+### Azure and Google cloud storage: MapReduce Intermediate Manifest Committer.
+
+Versions of the hadoop-mapreduce-core JAR shipped after September 2022 (3.3.5 and later)
+contain a committer optimized for performance and resilience on
+Azure ADLS Generation 2 and Google Cloud Storage.
+
+This committer, the "manifest committer" uses a manifest file to propagate
+directory listing information from the task committers to the job committer.
+These manifests can be written atomically, without relying on atomic directory rename,
+something GCS lacks.
+
+The job commmitter reads these manifests and will rename files from the task output
+directories directly into the destination directory, in parallel, with optional
+rate limiting to avoid throttling IO.
+This deliviers performance and scalability on the object stores.
+
+It is not critical for job correctness to use this with Azure storage; the
+classic FileOutputCommitter is safe there -however this new committer scales
Review Comment:
nit. `-however` -> `- however`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964734100
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -161,7 +204,16 @@ object PathOutputCommitProtocol {
val REJECT_FILE_OUTPUT_DEFVAL = false
/** Error string for tests. */
- private[cloud] val UNSUPPORTED: String = "PathOutputCommitProtocol does not support" +
+ private[cloud] val UNSUPPORTED: String = "PathOutputCommitter does not support" +
" dynamicPartitionOverwrite"
+ /**
+ * Stream Capabilities probe for spark dynamic partitioning compatibility.
+ */
+ private[cloud] val CAPABILITY_DYNAMIC_PARTITIONING = "mapreduce.job.committer.dynamic.partitioning"
+
+ /**
+ * Scheme prefix for per-filesystem scheme committers.
+ */
+ private[cloud] val OUTPUTCOMMITTER_FACTORY_SCHEME = "mapreduce.outputcommitter.factory.scheme"
Review Comment:
linked to the hadoop docs. we don't yet have a release with the manifest committer docs, but it is covered in https://github.com/apache/hadoop/blob/trunk/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/manifest_committer.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r965746274
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -125,22 +134,126 @@ class CommitterBindingSuite extends SparkFunSuite {
val instance = FileCommitProtocol.instantiate(
pathCommitProtocolClassname,
jobId, "file:///tmp", false)
-
val protocol = instance.asInstanceOf[PathOutputCommitProtocol]
assert("file:///tmp" === protocol.destination)
}
Review Comment:
done. went and compared the history and reverted any newline changes which expanded the diff
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] attilapiros commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
attilapiros commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r949452000
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -114,10 +108,38 @@ class PathOutputCommitProtocol(
// failures. Warn
logTrace(s"Committer $committer may not be tolerant of task commit failures")
}
+ } else {
+
Review Comment:
Nit: remove this line
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -114,10 +108,38 @@ class PathOutputCommitProtocol(
// failures. Warn
logTrace(s"Committer $committer may not be tolerant of task commit failures")
}
+ } else {
+
+ // if required other committers need to be checked for dynamic partition
+ // compatibility through a StreamCapabilities probe.
+ if (dynamicPartitionOverwrite) {
+ logDebug(s"Checking dynamic partition overwrite support in committer $committer")
Review Comment:
Nit: remove this debug line as the next line will be a DEBUG or an exception with all the information about the committer.
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -49,18 +49,27 @@ class CommitterBindingSuite extends SparkFunSuite {
* [[BindingParquetOutputCommitter]] committer bind to the schema-specific
* committer declared for the destination path? And that lifecycle events
* are correctly propagated?
+ * This only works with a hadoop build where BindingPathOutputCommitter
+ * does passthrough of stream capabilities, so check that first.
*/
test("BindingParquetOutputCommitter binds to the inner committer") {
+
Review Comment:
Nit: extra lines
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -140,6 +162,28 @@ class PathOutputCommitProtocol(
file.toString
}
+ /**
+ * Reject any requests for an absolute path file on a committer which
+ * is not compatible with it.
+ *
+ * @param taskContext task context
+ * @param absoluteDir final directory
+ * @param spec output filename
+ * @return a path string
+ * @throws UnsupportedOperationException if incompatible
+ */
+ override def newTaskTempFileAbsPath(
+ taskContext: TaskAttemptContext,
+ absoluteDir: String,
+ spec: FileNameSpec): String = {
+
+ if (supportsDynamicPartitions) {
+ super.newTaskTempFileAbsPath(taskContext, absoluteDir, spec)
+ } else {
+ throw new UnsupportedOperationException(s"Absolute output locations not supported" +
+ s" by committer $committer")
+ }
+ }
Review Comment:
Do we really need this? I mean in the `setupCommitter()` we already checked the `supportsDynamicPartitions` and an exception was thrown.
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -51,14 +51,7 @@ class PathOutputCommitProtocol(
jobId: String,
dest: String,
dynamicPartitionOverwrite: Boolean = false)
- extends HadoopMapReduceCommitProtocol(jobId, dest, false) with Serializable {
-
- if (dynamicPartitionOverwrite) {
- // until there's explicit extensions to the PathOutputCommitProtocols
- // to support the spark mechanism, it's left to the individual committer
- // choice to handle partitioning.
- throw new IOException(PathOutputCommitProtocol.UNSUPPORTED)
- }
+ extends HadoopMapReduceCommitProtocol(jobId, dest, dynamicPartitionOverwrite) with Serializable {
Review Comment:
Please update the comment above the class:
> dynamic partition overwrite is not supported, so that committers for stores which do not support rename will not get confused.
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -88,6 +81,7 @@ class PathOutputCommitProtocol(
logTrace(s"Setting up committer for path $destination")
committer = PathOutputCommitterFactory.createCommitter(destPath, context)
+
Review Comment:
Nit: remove this line
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -161,7 +205,18 @@ object PathOutputCommitProtocol {
val REJECT_FILE_OUTPUT_DEFVAL = false
/** Error string for tests. */
- private[cloud] val UNSUPPORTED: String = "PathOutputCommitProtocol does not support" +
+ private[cloud] val UNSUPPORTED: String = "PathOutputCommitter does not support" +
" dynamicPartitionOverwrite"
+
Review Comment:
Nit: excess line
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -161,7 +205,18 @@ object PathOutputCommitProtocol {
val REJECT_FILE_OUTPUT_DEFVAL = false
/** Error string for tests. */
- private[cloud] val UNSUPPORTED: String = "PathOutputCommitProtocol does not support" +
+ private[cloud] val UNSUPPORTED: String = "PathOutputCommitter does not support" +
" dynamicPartitionOverwrite"
+
+ /**
+ * Stream Capabilities probe for spark dynamic partitioning compatibility.
+ */
+ private[cloud] val CAPABILITY_DYNAMIC_PARTITIONING = "mapreduce.job.committer.dynamic.partitioning"
+
+
Review Comment:
Nit: extra line
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -114,10 +108,38 @@ class PathOutputCommitProtocol(
// failures. Warn
logTrace(s"Committer $committer may not be tolerant of task commit failures")
}
+ } else {
+
+ // if required other committers need to be checked for dynamic partition
+ // compatibility through a StreamCapabilities probe.
+ if (dynamicPartitionOverwrite) {
+ logDebug(s"Checking dynamic partition overwrite support in committer $committer")
+ if (supportsDynamicPartitions) {
+ logDebug(
+ s"Committer $committer has declared compatibility with dynamic partition overwrite")
+ } else {
+ throw new IOException(PathOutputCommitProtocol.UNSUPPORTED + ": " + committer)
+ }
+ }
}
+
Review Comment:
Nit: remove these 3 lines.
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -49,18 +49,27 @@ class CommitterBindingSuite extends SparkFunSuite {
* [[BindingParquetOutputCommitter]] committer bind to the schema-specific
* committer declared for the destination path? And that lifecycle events
* are correctly propagated?
+ * This only works with a hadoop build where BindingPathOutputCommitter
+ * does passthrough of stream capabilities, so check that first.
*/
test("BindingParquetOutputCommitter binds to the inner committer") {
+
+
val path = new Path("http://example/data")
val job = newJob(path)
val conf = job.getConfiguration
conf.set(MRJobConfig.TASK_ATTEMPT_ID, taskAttempt0)
conf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 1)
- StubPathOutputCommitterFactory.bind(conf, "http")
- val tContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+
+
+ StubPathOutputCommitterBinding.bindWithDynamicPartitioning(conf, "http")
+ val tContext: TaskAttemptContext = new TaskAttemptContextImpl(conf, taskAttemptId0)
+
+
Review Comment:
Nit: extra lines
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r950165502
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -51,14 +51,7 @@ class PathOutputCommitProtocol(
jobId: String,
dest: String,
dynamicPartitionOverwrite: Boolean = false)
- extends HadoopMapReduceCommitProtocol(jobId, dest, false) with Serializable {
-
- if (dynamicPartitionOverwrite) {
- // until there's explicit extensions to the PathOutputCommitProtocols
- // to support the spark mechanism, it's left to the individual committer
- // choice to handle partitioning.
- throw new IOException(PathOutputCommitProtocol.UNSUPPORTED)
- }
+ extends HadoopMapReduceCommitProtocol(jobId, dest, dynamicPartitionOverwrite) with Serializable {
Review Comment:
added an explanation of what now happens
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r965454664
##########
docs/cloud-integration.md:
##########
@@ -266,26 +271,123 @@ It has been tested with the most common formats supported by Spark.
mydataframe.write.format("parquet").save("s3a://bucket/destination")
```
-More details on these committers can be found in the latest Hadoop documentation.
+More details on these committers can be found in
+[the latest Hadoop documentation](https://hadoop.apache.org/docs/current/)
+with S3A committer detail covered in
+[Committing work to S3 with the S3A Committers](https://hadoop.apache.org/docs/current/hadoop-aws/tools/hadoop-aws/committers.html).
Note: depending upon the committer used, in-progress statistics may be
under-reported with Hadoop versions before 3.3.1.
+### Amazon EMR: the EMRFS S3-optimized committer
+
+Amazon EMR has its own S3-aware committers for parquet data.
+For instructions on use, see
+[the EMRFS S3-optimized committer](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-s3-optimized-committer.html)
+
+For implementation and performanc details, see
+["Improve Apache Spark write performance on Apache Parquet formats with the EMRFS S3-optimized committer"](https://aws.amazon.com/blogs/big-data/improve-apache-spark-write-performance-on-apache-parquet-formats-with-the-emrfs-s3-optimized-committer/
+
+
+### Azure and Google cloud storage: MapReduce Intermediate Manifest Committer.
+
+Versions of the hadoop-mapreduce-core JAR shipped after September 2022 (3.3.5 and later)
+contain a committer optimized for performance and resilience on
+Azure ADLS Generation 2 and Google Cloud Storage.
+
+This committer, the "manifest committer" uses a manifest file to propagate
+directory listing information from the task committers to the job committer.
+These manifests can be written atomically, without relying on atomic directory rename,
+something GCS lacks.
+
+The job commmitter reads these manifests and will rename files from the task output
+directories directly into the destination directory, in parallel, with optional
+rate limiting to avoid throttling IO.
+This deliviers performance and scalability on the object stores.
+
+It is not critical for job correctness to use this with Azure storage; the
+classic FileOutputCommitter is safe there -however this new committer scales
+better for large jobs with deep and wide directory trees.
+
+Because google GCS does not support atomic directory renaming,
Review Comment:
`google` -> `Google`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964637788
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -76,6 +78,16 @@ class CommitterBindingSuite extends SparkFunSuite {
assert(inner.jobCommitted, s"$inner job not committed")
parquet.abortJob(tContext, JobStatus.State.RUNNING)
assert(inner.jobAborted, s"$inner job not aborted")
+
+ val binding = new BindingPathOutputCommitter(path, tContext)
+ if (binding.isInstanceOf[StreamCapabilities]) {
Review Comment:
yes, because the new feature isn't in hadoop 3.3.4 or earlier....this test suite works with the current and future hadoop releases. the new feature needs the forthcoming hadoop branch-3.3. release
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] attilapiros commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
attilapiros commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964180597
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -161,7 +204,16 @@ object PathOutputCommitProtocol {
val REJECT_FILE_OUTPUT_DEFVAL = false
/** Error string for tests. */
- private[cloud] val UNSUPPORTED: String = "PathOutputCommitProtocol does not support" +
+ private[cloud] val UNSUPPORTED: String = "PathOutputCommitter does not support" +
" dynamicPartitionOverwrite"
+ /**
+ * Stream Capabilities probe for spark dynamic partitioning compatibility.
+ */
+ private[cloud] val CAPABILITY_DYNAMIC_PARTITIONING = "mapreduce.job.committer.dynamic.partitioning"
+
+ /**
+ * Scheme prefix for per-filesystem scheme committers.
+ */
+ private[cloud] val OUTPUTCOMMITTER_FACTORY_SCHEME = "mapreduce.outputcommitter.factory.scheme"
Review Comment:
Our documentation lack mentioning the `mapreduce.outputcommitter.factory.scheme` config and as I see this config will be more important than it was before (when it was just simply set to a single specific value in `SparkContext#fillMissingMagicCommitterConfsIfNeeded`).
What about extending the `cloud-integration.md` with some info about its role? Or just add a link to the Hadoop's documentation of this config?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r965740552
##########
hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala:
##########
@@ -161,7 +204,16 @@ object PathOutputCommitProtocol {
val REJECT_FILE_OUTPUT_DEFVAL = false
/** Error string for tests. */
- private[cloud] val UNSUPPORTED: String = "PathOutputCommitProtocol does not support" +
+ private[cloud] val UNSUPPORTED: String = "PathOutputCommitter does not support" +
" dynamicPartitionOverwrite"
+ /**
+ * Stream Capabilities probe for spark dynamic partitioning compatibility.
Review Comment:
fixed. interesting. i've switched my IDE to rendering scala/javadocs and it hid that. something for me to worry about. maybe I should turn it off again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun closed pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun closed pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
URL: https://github.com/apache/spark/pull/37468
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] steveloughran commented on pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
steveloughran commented on PR #37468:
URL: https://github.com/apache/spark/pull/37468#issuecomment-1239523356
added a new test case and updated the docs. I've not yet rebased it/merged it with your committer work, but the docs shouldn't clash. once this PR is in I will set my local build up to run your tests against s3 london
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r965456729
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -125,22 +134,126 @@ class CommitterBindingSuite extends SparkFunSuite {
val instance = FileCommitProtocol.instantiate(
pathCommitProtocolClassname,
jobId, "file:///tmp", false)
-
val protocol = instance.asInstanceOf[PathOutputCommitProtocol]
assert("file:///tmp" === protocol.destination)
}
Review Comment:
If you don't mind, could you revert the newline addition on the above test case, `test("committer protocol can be serialized and deserialized")`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] attilapiros commented on a diff in pull request #37468: [SPARK-40034][SQL] PathOutputCommitters to support dynamic partitions
Posted by GitBox <gi...@apache.org>.
attilapiros commented on code in PR #37468:
URL: https://github.com/apache/spark/pull/37468#discussion_r964177134
##########
hadoop-cloud/src/hadoop-3/test/scala/org/apache/spark/internal/io/cloud/CommitterBindingSuite.scala:
##########
@@ -76,6 +78,16 @@ class CommitterBindingSuite extends SparkFunSuite {
assert(inner.jobCommitted, s"$inner job not committed")
parquet.abortJob(tContext, JobStatus.State.RUNNING)
assert(inner.jobAborted, s"$inner job not aborted")
+
+ val binding = new BindingPathOutputCommitter(path, tContext)
+ if (binding.isInstanceOf[StreamCapabilities]) {
Review Comment:
Now the test is in the `hadoop-3` directory so do we still need this `if`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org