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 2021/06/21 15:48:54 UTC
[GitHub] [spark] cloud-fan opened a new pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
cloud-fan opened a new pull request #33002:
URL: https://github.com/apache/spark/pull/33002
<!--
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 message, please read the guideline first:
https://spark.apache.org/error-message-guidelines.html
-->
### 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.
-->
Currently, the batch and streaming file writers generate the file name a bit differently:
1. The batch file writer, generates the file name with partition ID and job id (which is a unique UUID)
2. The streaming file writer, generates the file name with partition ID and a fresh UUID
The reason for it is:
1. The file output committer for batch creates a unique staging directory for each task attempt, so partition ID is good enough to avoid file name collision, as concurrent task attempts (task retry, speculative tasks) are well handled by the staging directory.
2. The file output committe for streaming does not use staging directories. It writes files to the final path directly and uses a manifest file to track the committed files. Thus, partition ID is not sufficient to avoid file name collision. That's why we add a fresh UUID to the file name.
This PR proposes to unify the file name, by putting task attempt ID (which must be unique within the job) and the job ID in the file name.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Remove confusion when people try to understand how Spark generates file names. We can further refactor the code later to move the file name generation outside of the output committer.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No. The read side doesn't care about file name at all, and only care about how to list files. No backward compatibility issues.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
existing tests
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655506168
##########
File path: hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
##########
@@ -134,7 +134,7 @@ class PathOutputCommitProtocol(
val parent = dir.map {
d => new Path(workDir, d)
}.getOrElse(workDir)
- val file = new Path(parent, getFilename(taskContext, ext))
+ val file = new Path(parent, getFilename(ext))
Review comment:
I'm not sure if this change affects this new committer, but I think it should be a positive change. @steveloughran
##########
File path: hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
##########
@@ -134,7 +134,7 @@ class PathOutputCommitProtocol(
val parent = dir.map {
d => new Path(workDir, d)
}.getOrElse(workDir)
- val file = new Path(parent, getFilename(taskContext, ext))
+ val file = new Path(parent, getFilename(ext))
Review comment:
I'm not sure if this change affects this new committer, but I think it should be a positive change. The file name now use task attempt id instead of partition id, which is "more unique".
@steveloughran
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
A more aggressive way is to simply use a fresh UUID here, but I'm not sure if that's better. cc @zsxwing
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- val uuid = UUID.randomUUID.toString
- val filename = f"part-$split%05d-$uuid$ext"
Review comment:
Currently it does so to avoid file name collision, but I think it's overkill and we can use "task attempt id + job id" to avoid name collision as well, which is more consistent with the batch side.
It may also be useful to include the job id in the file name like the batch side does, so that people can see which files were written by the same job.
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
the value will be very different. For one query, the partition id always starts with 0. But task attempt id is unique within a spark application and won't be reset for a new query.
If we do want to keep the `part-00000` prefix for some reason, we can also apply the naming rule from streaming to batch. I don't know who will care about the final naming. The commit protocols I'm aware of only care about file listing.
--
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.
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] AmplabJenkins commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865212998
--
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.
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 change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
steveloughran commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657429293
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
part # is handy for a bit of blame assignment. FWIW the name "part" can be configured in ` "mapreduce.output.basename"`. Don't know if anyone does.
Now, the v2 committer whose lack of task commit idempotency is well known is only going be able to recover from a failure partway through task attempt commit if the second attempt creates files with the same name. This should not be a barrier to having better names as, well, it's still broken.
> But task attempt id is unique within a spark application and won't be reset for a new query.
this true? I think really need to understand differences between spark job, task ID and attempt IDs and the YARN ones, which as we know, have had duplicate job IDs until SPARK-33402.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
Review comment:
Specifically: Hadoop Task ID MUST be the same for all task attempts, so that committers can commit the output of more than one task attempt by renaming the the Task Attempt dir to output/_temporary/jobAttempt/taskID ; as only one task commit can do this (Assuming fs has atomic rename; google GCS doesn't), you get unique output. My WiP manifest committer creates a JSON manifest with task ID in the filename for the same reason: only one file can be committed by file rename (Atomic on GCS as well as azure).
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala
##########
@@ -147,9 +147,13 @@ class FileStreamSink(
if (batchId <= fileLog.getLatestBatchId().getOrElse(-1L)) {
logInfo(s"Skipping already committed batch $batchId")
} else {
+ // To avoid file name collision, we should generate a new job ID for every write job, instead
+ // of using batchId, as we may use the same batchId to write files again, if the streaming job
+ // fails and we restore from the checkpoint.
+ val jobId = java.util.UUID.randomUUID().toString
Review comment:
One change of SPARK-33402 was including some timestamp/version info. That's potentially quite handy later just to see when things were created/order
##########
File path: hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
##########
@@ -134,7 +134,7 @@ class PathOutputCommitProtocol(
val parent = dir.map {
d => new Path(workDir, d)
}.getOrElse(workDir)
- val file = new Path(parent, getFilename(taskContext, ext))
+ val file = new Path(parent, getFilename(ext))
Review comment:
Commit protocols MUST NOT contain any assumptions about filenames. It would be silly.
Well, almost not. try creating a file with .pending or .pendingset in the magic committer and it'd be very confused. (Maybe we should change that to something really obscure...)
--
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.
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] cloud-fan commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865145708
cc @viirya @squito @HeartSaVioR
--
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.
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] AmplabJenkins removed a comment on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865212998
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44612/
--
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.
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] SparkQA commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865204035
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44612/
--
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.
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] AmplabJenkins removed a comment on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865286344
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140084/
--
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.
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] viirya commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655911943
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
Previously it uses task id after `part-`, now this is `taskAttemptId`. Is it still the same format as before, e.g. `part-xxxxx-`?
--
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.
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] viirya commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655911943
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
Previously it uses task id after `part-`, now this is `taskAttemptId`. Is it still the same format as before, e.g. `part-xxxxx-`?
--
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.
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] SparkQA removed a comment on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865176381
**[Test build #140084 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140084/testReport)** for PR 33002 at commit [`715ce3c`](https://github.com/apache/spark/commit/715ce3c22a4934c23a958e0f9b7351a8e72a083f).
--
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.
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] SparkQA commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865273823
**[Test build #140084 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140084/testReport)** for PR 33002 at commit [`715ce3c`](https://github.com/apache/spark/commit/715ce3c22a4934c23a958e0f9b7351a8e72a083f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655508695
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
A more aggressive way is to simply use a fresh UUID here, but I'm not sure if that's better. cc @zsxwing
--
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.
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] zsxwing commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
zsxwing commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655627123
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- val uuid = UUID.randomUUID.toString
- val filename = f"part-$split%05d-$uuid$ext"
Review comment:
> 2\. The file output committe for streaming does not use staging directories. It writes files to the final path directly and uses a manifest file to track the committed files. Thus, partition ID is not sufficient to avoid file name collision. That's why we add a fresh UUID to the file name.
Could you explain this? Currently `ManifestFileCommitProtocol` should always pick up a new uuid for each file.
--
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.
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] github-actions[bot] commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-932837574
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657620317
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
Review comment:
> Hadoop Task ID MUST be the same for all task attempts
This doesn't change: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala#L268
This PR is only to unify the file name generated by the builtin Spark file commit protocol, and doesn't change anything Hadoop Job/Task setting.
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655506168
##########
File path: hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
##########
@@ -134,7 +134,7 @@ class PathOutputCommitProtocol(
val parent = dir.map {
d => new Path(workDir, d)
}.getOrElse(workDir)
- val file = new Path(parent, getFilename(taskContext, ext))
+ val file = new Path(parent, getFilename(ext))
Review comment:
I'm not sure if this change affects this new committer, but I think it should be a positive change. The file name now use task attempt id instead of partition id, which is "more unique".
@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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655950782
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
the value will be very different. For one query, the partition id always starts with 0. But task attempt id is unique within a spark application and won't be reset for a new query.
If we do want to keep the `part-00000` prefix for some reason, we can also apply the naming rule from streaming to batch. I don't know who will care about the final naming. The commit protocols I'm aware of only care about file listing.
--
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.
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] AmplabJenkins removed a comment on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865212998
--
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.
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] AmplabJenkins commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865212998
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44612/
--
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.
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 #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
steveloughran commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-867119415
ok, this gets into a world of fun; thanks for mentioning me.
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657620317
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
Review comment:
> Hadoop Task ID MUST be the same for all task attempts
This doesn't change: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala#L268
This PR is only to unify the file name generated by the builtin Spark file commit protocol, and doesn't change anything in Hadoop Job/Task setting.
--
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.
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] SparkQA commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865176381
**[Test build #140084 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140084/testReport)** for PR 33002 at commit [`715ce3c`](https://github.com/apache/spark/commit/715ce3c22a4934c23a958e0f9b7351a8e72a083f).
--
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.
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] SparkQA removed a comment on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865176381
**[Test build #140084 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140084/testReport)** for PR 33002 at commit [`715ce3c`](https://github.com/apache/spark/commit/715ce3c22a4934c23a958e0f9b7351a8e72a083f).
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655506168
##########
File path: hadoop-cloud/src/hadoop-3/main/scala/org/apache/spark/internal/io/cloud/PathOutputCommitProtocol.scala
##########
@@ -134,7 +134,7 @@ class PathOutputCommitProtocol(
val parent = dir.map {
d => new Path(workDir, d)
}.getOrElse(workDir)
- val file = new Path(parent, getFilename(taskContext, ext))
+ val file = new Path(parent, getFilename(ext))
Review comment:
I'm not sure if this change affects this new committer, but I think it should be a positive change. @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.
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] viirya commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657691597
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
Review comment:
same, taskId -> taskAttemptId?
--
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.
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] AmplabJenkins commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865286344
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140084/
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655651456
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- val uuid = UUID.randomUUID.toString
- val filename = f"part-$split%05d-$uuid$ext"
Review comment:
Currently it does so to avoid file name collision, but I think it's overkill and we can use "task attempt id + job id" to avoid name collision as well, which is more consistent with the batch side.
It may also be useful to include the job id in the file name like the batch side does, so that people can see which files were written by the same job.
--
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.
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] viirya commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657690836
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
Review comment:
taskId -> taskAttemptId?
--
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.
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] github-actions[bot] closed pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #33002:
URL: https://github.com/apache/spark/pull/33002
--
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657619489
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala
##########
@@ -147,9 +147,13 @@ class FileStreamSink(
if (batchId <= fileLog.getLatestBatchId().getOrElse(-1L)) {
logInfo(s"Skipping already committed batch $batchId")
} else {
+ // To avoid file name collision, we should generate a new job ID for every write job, instead
+ // of using batchId, as we may use the same batchId to write files again, if the streaming job
+ // fails and we restore from the checkpoint.
+ val jobId = java.util.UUID.randomUUID().toString
Review comment:
This is the job id for spark file commit protocol. In Hadoop `JobId`, we do append the timestamp info: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala#L266
But that's a different story.
--
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.
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] SparkQA commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865176381
--
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.
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] zsxwing commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
zsxwing commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r655627123
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ManifestFileCommitProtocol.scala
##########
@@ -113,12 +113,15 @@ class ManifestFileCommitProtocol(jobId: String, path: String)
override def newTaskTempFile(
taskContext: TaskAttemptContext, dir: Option[String], ext: String): String = {
- // The file name looks like part-r-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- val uuid = UUID.randomUUID.toString
- val filename = f"part-$split%05d-$uuid$ext"
Review comment:
> 2\. The file output committe for streaming does not use staging directories. It writes files to the final path directly and uses a manifest file to track the committed files. Thus, partition ID is not sufficient to avoid file name collision. That's why we add a fresh UUID to the file name.
Could you explain this? Currently `ManifestFileCommitProtocol` should always pick up a new uuid for each file.
--
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.
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] cloud-fan commented on a change in pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #33002:
URL: https://github.com/apache/spark/pull/33002#discussion_r657610207
##########
File path: core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
##########
@@ -152,12 +153,21 @@ class HadoopMapReduceCommitProtocol(
tmpOutputPath
}
- protected def getFilename(taskContext: TaskAttemptContext, ext: String): String = {
- // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003-c000.parquet
- // Note that %05d does not truncate the split number, so if we have more than 100000 tasks,
+ protected def getFilename(ext: String): String = {
+ // Use the Spark task attempt ID which is unique within the write job, so that file writes never
+ // collide if the file name also includes job ID. The Hadoop task id is equivalent to Spark's
+ // partitionId, which is not unique within the write job, for cases like task retry or
+ // speculative tasks.
+ // NOTE: this is not necessary for certain Hadoop output committers, as they will create a
+ // unique staging directory for each task attempt, so we don't need to worry about file name
+ // collision between different task attempts, and using Hadoop task ID/Spark partition ID is
+ // also fine. For extra safety and consistency with the streaming side, we always use the
+ // Spark task attempt ID here.
+ val taskId = TaskContext.get.taskAttemptId()
+ // The file name looks like part-00000-2dd664f9-d2c4-4ffe-878f-c6c70c1fb0cb_00003.gz.parquet
+ // Note that %05d does not truncate the taskId, so if we have more than 100000 tasks,
// the file name is fine and won't overflow.
- val split = taskContext.getTaskAttemptID.getTaskID.getId
- f"part-$split%05d-$jobId$ext"
+ f"part-$taskId%05d-$jobId$ext"
Review comment:
Spark has: job id -> stage id -> partition ID
job id is simply a UUID
stage id is an integer starting from 0, globally unique within the spark application
partition ID is an integer starting from 0, unique within a stage
Eash task does not only have partition ID, but also has attempt ID, which is an integer starting from 0, globally unique within the spark application. There is also an attempt number, which starts from 0 and increases by one for each attempt of this task.
--
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.
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] cloud-fan commented on pull request #33002: [SPARK-35843][SQL] Unify the file name between batch and streaming file writers
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #33002:
URL: https://github.com/apache/spark/pull/33002#issuecomment-865145708
cc @viirya @squito @HeartSaVioR
--
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.
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