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/04/05 06:14:05 UTC
[GitHub] [spark] CHENXCHEN opened a new pull request, #36070: [SPARK-32838][CORE] Fix rename and delete files with different filesystem
CHENXCHEN opened a new pull request, #36070:
URL: https://github.com/apache/spark/pull/36070
### What changes were proposed in this pull request?
When we use a partition table, if the filesystem of a partition is different from the filesystem of the table location,
we will get an exception：`java.lang.IllegalArgumentException: Wrong FS: s3a://path/to/spark3_snap/dt=2020-09-10, expected: hdfs://cluster`,
because `HadoopMapReduceCommitProtocol` will use the filesystem of the table location to operate the file
For example, the following SQL will cause the above exception:
```sql
CREATE TABLE `spark3_snap`( `id` string) PARTITIONED BY (`dt` string)
STORED AS ORC LOCATION 'hdfs://path/to/spark3_snap';
-- The file system of the partition location is different from the filesystem of the table location,
-- one is S3A, the other is HDFS
alter table tmp.spark3_snap add partition (dt='2020-09-10')
LOCATION 's3a://path/to/spark3_snap/dt=2020-09-10';
insert overwrite table tmp.spark3_snap partition(dt)
select '10' id, '2020-09-09' dt
union
select '20' id, '2020-09-10' dt
;
```
### Why are the changes needed?
We cannot operate on partitions with different from filesystem of table partition location
### Does this PR introduce _any_ user-facing change?
Yes, before this PR, an exception will be reported when the user operates a filesystem of partition location different from the filesystem of table location. After this PR, it will be processed as needed.
### How was this patch tested?
Manual testing
--
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] github-actions[bot] commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1320999302
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] steveloughran commented on a diff in pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #36070:
URL: https://github.com/apache/spark/pull/36070#discussion_r943432160
##########
core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala:
##########
@@ -194,19 +196,33 @@ class HadoopMapReduceCommitProtocol(
if (hasValidPath) {
val (allAbsPathFiles, allPartitionPaths) =
taskCommits.map(_.obj.asInstanceOf[(Map[String, String], Set[String])]).unzip
- val fs = stagingDir.getFileSystem(jobContext.getConfiguration)
+ val hadoopConf = jobContext.getConfiguration
+ val fs = stagingDir.getFileSystem(hadoopConf)
val filesToMove = allAbsPathFiles.foldLeft(Map[String, String]())(_ ++ _)
logDebug(s"Committing files staged for absolute locations $filesToMove")
val absParentPaths = filesToMove.values.map(new Path(_).getParent).toSet
if (dynamicPartitionOverwrite) {
logDebug(s"Clean up absolute partition directories for overwriting: $absParentPaths")
- absParentPaths.foreach(fs.delete(_, true))
+ absParentPaths.foreach(path => path.getFileSystem(hadoopConf)
+ .delete(path, true))
}
logDebug(s"Create absolute parent directories: $absParentPaths")
- absParentPaths.foreach(fs.mkdirs)
+ absParentPaths.foreach(path => path.getFileSystem(hadoopConf).mkdirs(path))
for ((src, dst) <- filesToMove) {
- if (!fs.rename(new Path(src), new Path(dst))) {
+ val srcPath = new Path(src)
+ val dstPath = new Path(dst)
+ val srcFs = srcPath.getFileSystem(hadoopConf)
+ val dstFs = dstPath.getFileSystem(hadoopConf)
+ // Copying files across different file systems
+ if (needCopy(srcPath, dstPath, srcFs, dstFs)) {
+ if (!FileUtil.copy(srcFs, srcFs.listStatus(srcPath).map(_.getPath), dstFs, dstPath,
Review Comment:
you may want to think about parallelizing the copy, as for each file it is now going to take time proportional to data.length/(download_bandwidth+upload_bandwidth)
shame copy returns false sometimes; looks like it is only if mkdirs() on the dest or delete(src) fails.
--
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 diff in pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36070:
URL: https://github.com/apache/spark/pull/36070#discussion_r844771210
##########
core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala:
##########
@@ -194,19 +196,33 @@ class HadoopMapReduceCommitProtocol(
if (hasValidPath) {
val (allAbsPathFiles, allPartitionPaths) =
taskCommits.map(_.obj.asInstanceOf[(Map[String, String], Set[String])]).unzip
- val fs = stagingDir.getFileSystem(jobContext.getConfiguration)
+ val hadoopConfiguration = jobContext.getConfiguration
Review Comment:
```suggestion
val hadoopConf = jobContext.getConfiguration
```
--
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 pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1091212615
Cross-file-system table writing sounds like a big feature to me. Currently, Spark fails to write so this feature is not supported yet, but I'm wondering what's the best way to do it, e.g. shall we put the staging dir in the same file system of the target path?
cc @AngersZhuuuu @yaooqinn
--
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] AmplabJenkins commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1088457342
Can one of the admins verify this patch?
--
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 #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
steveloughran commented on code in PR #36070:
URL: https://github.com/apache/spark/pull/36070#discussion_r1029595674
##########
core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala:
##########
@@ -194,19 +196,33 @@ class HadoopMapReduceCommitProtocol(
if (hasValidPath) {
val (allAbsPathFiles, allPartitionPaths) =
taskCommits.map(_.obj.asInstanceOf[(Map[String, String], Set[String])]).unzip
- val fs = stagingDir.getFileSystem(jobContext.getConfiguration)
+ val hadoopConf = jobContext.getConfiguration
+ val fs = stagingDir.getFileSystem(hadoopConf)
val filesToMove = allAbsPathFiles.foldLeft(Map[String, String]())(_ ++ _)
logDebug(s"Committing files staged for absolute locations $filesToMove")
val absParentPaths = filesToMove.values.map(new Path(_).getParent).toSet
if (dynamicPartitionOverwrite) {
logDebug(s"Clean up absolute partition directories for overwriting: $absParentPaths")
- absParentPaths.foreach(fs.delete(_, true))
+ absParentPaths.foreach(path => path.getFileSystem(hadoopConf)
+ .delete(path, true))
}
logDebug(s"Create absolute parent directories: $absParentPaths")
- absParentPaths.foreach(fs.mkdirs)
+ absParentPaths.foreach(path => path.getFileSystem(hadoopConf).mkdirs(path))
for ((src, dst) <- filesToMove) {
- if (!fs.rename(new Path(src), new Path(dst))) {
+ val srcPath = new Path(src)
+ val dstPath = new Path(dst)
+ val srcFs = srcPath.getFileSystem(hadoopConf)
+ val dstFs = dstPath.getFileSystem(hadoopConf)
+ // Copying files across different file systems
+ if (needCopy(srcPath, dstPath, srcFs, dstFs)) {
+ if (!FileUtil.copy(srcFs, srcFs.listStatus(srcPath).map(_.getPath), dstFs, dstPath,
Review Comment:
can i highlight something i've noticed here, that copy() command stos on src read() returning -1, without doing any checks to validate file length, not great.
--
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] github-actions[bot] closed pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
URL: https://github.com/apache/spark/pull/36070
--
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] CHENXCHEN commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
CHENXCHEN commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1091389301
> Current DS insert only support write staging dir for dynamic partition overwrite, this pr's case seems is to use hive serde(since hive serde support config staging dir use different file system). And spark's commit protocol not support different file system for staging dir.
>
> Add support for different file system have to consider a lot, you can check the point mentioned in #33828
>
> Also I am writing a new build in commit protocol #36056, it's behavior like hive, and it make all overwrite use a staging dir in the same file system of the target path.
Yes, this pr's case is to use hive serde(hive serde support config staging dir use different file system)
#33828 and #36056 If we have multiple filesystems that are different from the staging dir filesystem, we still have exceptions.
--
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] CHENXCHEN commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
CHENXCHEN commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1088339027
cc @cloud-fan could you help take a look when you have time? Thanks.
--
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] CHENXCHEN commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
CHENXCHEN commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1091300982
Staging Dir is generated based on the table location
If we specify that the generated partition location must be placed under the table location, we need:
1. Find the partition locations that are different from the table locations, change their locations, and then delete the old locations at [InsertIntoHadoopFsRelationCommand.scala](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala#L142)
2. Partition locations that are different from the table location need to be passed into the commiter as new task file at [FileFormatDataWriter.scala](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatDataWriter.scala#L292)
Hive's approach is to keep the path to the partition location and move files across file systems.[Hive.java](https://github.com/apache/hive/blob/rel/release-2.3.9/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java#L3156)
Whether our behavior should be consistent with that of hive?
--
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] CHENXCHEN commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
CHENXCHEN commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1091329593
Do we need a wrapper file system to handle all the files in spark, including cross file system operations?
It sounds like a big change...
--
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] AngersZhuuuu commented on pull request #36070: [SPARK-31675][CORE] Fix rename and delete files with different filesystem
Posted by GitBox <gi...@apache.org>.
AngersZhuuuu commented on PR #36070:
URL: https://github.com/apache/spark/pull/36070#issuecomment-1091328814
Current DS insert only support write staging dir for dynamic partition overwrite, this pr's case seems is to use hive serde(since hive serde support config staging dir use different file system). And spark's commit protocol not support different file system for staging dir.
Add support for different file system have to consider a lot, you can check the point mentioned in https://github.com/apache/spark/pull/33828
Also I am writing a new build in commit protocol https://github.com/apache/spark/pull/36056, it's behavior like hive, and it make all overwrite use a staging dir in the same file system of the target path.
--
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