You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "sandip-db (via GitHub)" <gi...@apache.org> on 2025/06/15 22:26:59 UTC
[PR] [SPARK-52482][SQL] ZStandard support for file data source reader [spark]
sandip-db opened a new pull request, #51182:
URL: https://github.com/apache/spark/pull/51182
### What changes were proposed in this pull request?
The current Hadoop version in Spark hasn’t been compiled with Zstd because of which ZSTD compressed json, csv, txt and xml files cannot be read. Spark has a built-in `ZStdCompressionCodec`, which is used for compressing shuffled data and other use cases.
This PR adds support for ZStandard decompression in file data source readers using the built-in `ZStdCompressionCodec`. It also add support for some non-standard extensions like `.gzip` and `.zstd`.
The first commit in this PR is a vanilla copy of Hadoop's `LineRecordReader`. This was needed to add support for non-Hadoop based codec. Reviewers need not review this commit.
Following config have been added to enable/disable changes in this PR:
- `spark.sql.execution.datasources.hadoopLineRecordReader.enabled`: Setting this SQLCOnf to false will result in using Hadoop's `LineRecordReader`. Otherwise, use the inlined `HadoopLineRecordReader` by default.
- `spark.io.zStandard.enabled`: Enable Spark's `ZStdCompressionCodec` for file data source reader. Default: `true`
### Why are the changes needed?
Same as above.
### Does this PR introduce _any_ user-facing change?
Yes, Adds support for ZSTD decompression in file data source readers.
### How was this patch tested?
Added new unit tests
### Was this patch authored or co-authored using generative AI tooling?
No
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "jackierwzhang (via GitHub)" <gi...@apache.org>.
jackierwzhang commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2150952529
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/HadoopLineRecordReader.java:
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import scala.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.Seekable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.CodecPool;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.SplitCompressionInputStream;
+import org.apache.hadoop.io.compress.SplittableCompressionCodec;
+import org.apache.hadoop.io.compress.Decompressor;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.hadoop.mapreduce.lib.input.SplitLineReader;
+import org.apache.hadoop.mapreduce.lib.input.CompressedSplitLineReader;
+import org.apache.hadoop.mapreduce.lib.input.UncompressedSplitLineReader;
+import org.apache.hadoop.util.functional.FutureIO;
+import org.apache.spark.internal.SparkLogger;
+import org.apache.spark.internal.SparkLoggerFactory;
+import org.apache.spark.io.HadoopCodecStreams;
+
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_SPLIT_END;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_SPLIT_START;
+
+/**
+ * Inlined from Hadoop's LineRecordReader to add support for passing compression option
+ * and also add support for other codecs like ZSTD.
+ */
+
+/**
+ * Treats keys as offset in file and value as line.
+ */
+public class HadoopLineRecordReader extends RecordReader<LongWritable, Text> {
Review Comment:
Could you describe the changes with more details for this inlined library? Maybe add inline comments to the exact lines that are altered
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2994697870
thanks, merging to master!
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2974887472
Hadoop has built-in `org.apache.hadoop.io.compress.ZStandardCodec`, but it requires compilation with the native library, I think the right direction is to migrate it to zstd-jni, like other codecs, see HADOOP-17292 (lz4), HADOOP-17125 (snappy), HADOOP-17825 (gzip).
Even if you don't want to touch the Hadoop code, this PR approach looks too overkill, Hadoop provides `io.compression.codecs.CompressionCodec` to allow implementing custom codecs, implementing a `org.apache.spark.xxx.SparkZstdCompressionCodec` and rewriting `io.compression.codecs` should work.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #51182: [SPARK-52482][SQL][CORE] ZStandard support for file data source reader
URL: https://github.com/apache/spark/pull/51182
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2982186085
> The inlining of Hadoop's LineRecordReader would be needed regardless of ZSTD support because of a follow-up change that I am working on, which would add allow users to specify compression type using file data source option.
@pan3793 does Hadoop's `LineRecordReader` allow us to specify the compression at the session level without forking the code?
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "jackierwzhang (via GitHub)" <gi...@apache.org>.
jackierwzhang commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2150968450
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.io
+
+import java.io.InputStream
+
+import scala.collection.Seq
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.compress._
+
+import org.apache.spark.{SparkConf, SparkEnv}
+import org.apache.spark.internal.config
+import org.apache.spark.io.{CompressionCodec => SparkCompressionCodec}
+
+object HadoopCodecStreams {
+ val ZSTD_EXTENSIONS = Seq(".zstd", ".zst")
+
+ // get codec based on file name extension
+ def getDecompressionCodec(
+ config: Configuration,
+ file: Path): Option[CompressionCodec] = {
+ val factory = new CompressionCodecFactory(config)
+ Option(factory.getCodec(file)).orElse {
+ // Try some non-standards extensions for Zstandard and Gzip
+ file.getName.toLowerCase() match {
+ case name if name.endsWith(".zstd") =>
+ Option(factory.getCodecByName(classOf[ZStandardCodec].getName))
+ case name if name.endsWith(".gzip") =>
+ Option(factory.getCodecByName(classOf[GzipCodec].getName))
+ case _ => None
+ }
+ }
+ }
+
+ def createZstdInputStream(
+ file: Path,
+ inputStream: InputStream): Option[InputStream] = {
+ val fileName = file.getName.toLowerCase()
+
+ if (ZSTD_EXTENSIONS.exists(fileName.endsWith) && isSparkZstdCodecEnabled) {
+ Some(
+ SparkCompressionCodec
+ .createCodec(sparkConf, SparkCompressionCodec.ZSTD)
+ .compressedInputStream(inputStream)
+ )
+ } else {
+ None
+ }
+ }
+
+ def createInputStream(config: Configuration, file: Path): InputStream = {
+ val fs = file.getFileSystem(config)
+ val inputStream: InputStream = fs.open(file)
+
+ getDecompressionCodec(config, file)
+ .map { codec =>
+ try {
+ codec.createInputStream(inputStream)
+ } catch {
+ case e: Exception =>
+ createZstdInputStream(file, inputStream).getOrElse(throw e)
Review Comment:
This logic is subtle, why is this fallback required?
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-3026537126
@sandip-db I just found an incompatible issue between zstd-jni and Hadoop native zstd reported by the HBase community. [HBASE-27706](https://issues.apache.org/jira/browse/HBASE-27706), not sure if it will affect Spark cases.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-3026618691
@sandip-db I know that, but since this is related to storage, we should be careful if there are incompatible possibilities
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2984451057
> for "no extensions" compressed text files, I'm not sure if this is a valid use case(see my last comment)
@pan3793 While uncommon, we come across users who have compressed files without any extension. Most recently, an user had to rename a billion gzipped compressed files in S3 because it didn't have any extension.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2982827433
> does Hadoop's LineRecordReader allow us to specify the compression at the session level without forking the code?
@cloud-fan as @sandip-db said, Hadoop's `LineRecordReader` relies on the filename suffix to choose the decompressor, I think this is a well-known behavior for processing text files, because unlike binary files (e.g. Parquet/ORC) which has metadata in footer to mark the codec used by each page/chunk inside binary file, compression is applied to the whole text file.
how do you define the behavior of "specify the compression at the session level"? always respect session conf and ignore filename suffix? or fallback to use codec suggested by session conf when something goes wrong?
also, please be careful with that Hadoop codec may have different behavior with Spark/Unix tool codec, for example, HADOOP-12990(lz4)
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2982202346
> does Hadoop's LineRecordReader allow us to specify the compression at the session level without forking the code?
@cloud-fan Its possible to pass different codecs via `io.compression.codecs.CompressionCodec` Hadoop conf, but Hadoop will still use file name extension to choose codec from that list. Users' files may not have a standard extension to begin with.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2153372192
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -5949,6 +5949,17 @@ object SQLConf {
.createWithDefault(2)
}
+ val HADOOP_LINE_RECORD_READER_ENABLED =
Review Comment:
```suggestion
val HADOOP_LINE_RECORD_READER_ENABLED =
```
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2151149487
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.io
+
+import java.io.InputStream
+
+import scala.collection.Seq
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.compress._
+
+import org.apache.spark.{SparkConf, SparkEnv}
+import org.apache.spark.internal.config
+import org.apache.spark.io.{CompressionCodec => SparkCompressionCodec}
+
+object HadoopCodecStreams {
+ val ZSTD_EXTENSIONS = Seq(".zstd", ".zst")
+
+ // get codec based on file name extension
+ def getDecompressionCodec(
+ config: Configuration,
+ file: Path): Option[CompressionCodec] = {
+ val factory = new CompressionCodecFactory(config)
+ Option(factory.getCodec(file)).orElse {
+ // Try some non-standards extensions for Zstandard and Gzip
+ file.getName.toLowerCase() match {
+ case name if name.endsWith(".zstd") =>
+ Option(factory.getCodecByName(classOf[ZStandardCodec].getName))
+ case name if name.endsWith(".gzip") =>
+ Option(factory.getCodecByName(classOf[GzipCodec].getName))
+ case _ => None
+ }
+ }
+ }
+
+ def createZstdInputStream(
+ file: Path,
+ inputStream: InputStream): Option[InputStream] = {
+ val fileName = file.getName.toLowerCase()
+
+ if (ZSTD_EXTENSIONS.exists(fileName.endsWith) && isSparkZstdCodecEnabled) {
+ Some(
+ SparkCompressionCodec
+ .createCodec(sparkConf, SparkCompressionCodec.ZSTD)
+ .compressedInputStream(inputStream)
+ )
+ } else {
+ None
+ }
+ }
+
+ def createInputStream(config: Configuration, file: Path): InputStream = {
+ val fs = file.getFileSystem(config)
+ val inputStream: InputStream = fs.open(file)
+
+ getDecompressionCodec(config, file)
+ .map { codec =>
+ try {
+ codec.createInputStream(inputStream)
+ } catch {
+ case e: Exception =>
+ createZstdInputStream(file, inputStream).getOrElse(throw e)
Review Comment:
createInputStream may fail with an Exception for ZSTD if hadoop is not already compiled with ZSTD support. In that case, we try to use Spark's Zstandard codec if enabled. Added a comment to clarify this.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2986645298
Instead of relying on the filename suffix or Spark session conf to choose the codec, I wonder if the Magic Number was considered? For example,
```
$ file unit-tests.log
unit-tests.log: ASCII text, with very long lines (388)
$ file unit-tests.log.gz
unit-tests.log.gz: gzip compressed data, was "unit-tests.log", last modified: Mon Apr 21 13:03:04 2025, from Unix, original size modulo 2^32 11024393
$ file unit-tests.log.zst
unit-tests.log.zst: Zstandard compressed data (v0.8+), Dictionary ID: None
$ cp unit-tests.log.gz unit-tests.log.gz.foo
$ file unit-tests.log.gz.foo
unit-tests.log.gz.foo: gzip compressed data, was "unit-tests.log", last modified: Mon Apr 21 13:03:04 2025, from Unix, original size modulo 2^32 11024393
$ file unit-tests.log.zst.bar
unit-tests.log.zst.bar: Zstandard compressed data (v0.8+), Dictionary ID: None
```
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2986630488
@sandip-db TBH, the current approach (control flow is based on try-catch exception) seems too hacky, and I'd like to see more detailed designs of your next steps. (I don't think the below question got answered.)
> how do you define the behavior of "specify the compression at the session level"? always respect session conf and ignore filename suffix? or fallback to use codec suggested by session conf when something goes wrong?
> This PR just adds decompression support.
You still need to ensure that Spark's zstd codec is compatible with Hadoop's implementation. I have experienced using the AirCompressor LZO codec to decompress the files written via hadoop-lzo may randomly get corrupt content with no errors.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "jackierwzhang (via GitHub)" <gi...@apache.org>.
jackierwzhang commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2150961678
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.io
+
+import java.io.InputStream
+
+import scala.collection.Seq
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.compress._
+
+import org.apache.spark.{SparkConf, SparkEnv}
+import org.apache.spark.internal.config
+import org.apache.spark.io.{CompressionCodec => SparkCompressionCodec}
+
+object HadoopCodecStreams {
+ val ZSTD_EXTENSIONS = Seq(".zstd", ".zst")
+
+ // get codec based on file name extension
+ def getDecompressionCodec(
+ config: Configuration,
+ file: Path): Option[CompressionCodec] = {
+ val factory = new CompressionCodecFactory(config)
+ Option(factory.getCodec(file)).orElse {
+ // Try some non-standards extensions for Zstandard and Gzip
+ file.getName.toLowerCase() match {
Review Comment:
nit: `Locale.ROOT`
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "rcalme (via GitHub)" <gi...@apache.org>.
rcalme commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-3304818639
@sandip-db is there any plan for future work to allow for this read pattern?
> User should be able to specify the compression type in their query using data source reader option. For example:
spark.read.option("compression", "gzip").json(path)
I'm not seeing any other currently supported way to achieve such a result.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-3026608102
@pan3793 Thanks for pointing it out. If native hadoop or user provided zstd codec is available, we will [use that](https://github.com/apache/spark/pull/51182/files#diff-8bf2be281511318ebeb1e5d306a9f266d78fa0f78f2e8134760acf4ef084eafcR84-R91) instead of Spark's zstd-JNI based decompression.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2978639268
> Even if you don't want to touch the Hadoop code, this PR approach looks too overkill, Hadoop provides io.compression.codecs.CompressionCodec to allow implementing custom codecs, implementing a org.apache.spark.xxx.SparkZstdCompressionCodec and configuring io.compression.codecs should work.
@pan3793 Thanks for your comment. The PR adds less than 50 lines of code to re-use Spark's `ZStdCompressionCodec` for file data source. The inlining of Hadoop's `LineRecordReader` would be needed regardless of ZSTD support because of a follow-up change that I am working on, which would add allow users to specify compression type using file data source option. There are users who have files with either non-standard extensions or no extensions at all. Hadoop's way of determining codec based on file name extension is forcing them to rename their files.
Implementing `CompressionCodec` and `Decompressor` seems unnecessary at this point. At some point, I expect Spark will upgrade to Hadoop with native ZSTD support and then the code will start to use that automatically instead of Spark's `ZStdCompressionCodec`.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2986808266
@pan3793 Thanks for your input.
> how do you define the behavior of "specify the compression at the session level"? always respect session conf and ignore filename suffix? or fallback to use codec suggested by session conf when something goes wrong?
User should be able to specify the compression type in their query using data source reader option. For example:
`spark.read.option("compression", "gzip").json(path)`
If this option is specified, Spark will always use the compression type specified by the user. There are some other alternatives available like first use the file path extension or use the magic number (used by the file utility) to determine the codec type.
> Instead of relying on the filename suffix or Spark session conf to choose the codec, I wonder if the Magic Number was considered? For example, the file command can correctly recognize the file codec even without standard filename extension
I agree and I wonder why Hadoop didn't do this in the first place. There will be some performance penalty for examining the magic number and reopening the file input stream with appropriate codec.
Having said that, we can take the discussion of the compression option and the use of magic number to my next PR.
The current PR is about adding ZSTD decompression support. I would appreciate if we can close this first.
The try-catch logic has been used to pick codec in the following order:
- User provided ZSTD codec via `io.compression.codecs.CompressionCodec` Hadoop Conf
- Hadoop's default ZSTD codec
- Spark's `ZStdCompressionCodec`.
For compatibility check, I have added some more tests that reads files compressed with ZSTD in ubuntu (version: 1.4.4+dfsg-3ubuntu0.1).
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2986934134
OK, it's fair enough to fork Hadoop's `LineRecordReader` and use try-catch logic for codec fallback given your flexible extending design.
> There will be some performance penalty for examining the magic number and reopening the file input stream with appropriate codec.
Have a wrapper Codec to look ahead Magic Number and transfer the InputStream to the concrete Codec should eliminate the re-open cost, anyway, this is about implementation details and can be discussed later.
> For compatibility check, I have added some more tests that reads files compressed with ZSTD in ubuntu (version: 1.4.4+dfsg-3ubuntu0.1).
I think you should at least test reading zstd text file written by Hadoop
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2987352849
> I think you should at least test reading zstd text file written by Hadoop
Added a test scenario with a file compressed using Hadoop native ZSTD codec.
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "jackierwzhang (via GitHub)" <gi...@apache.org>.
jackierwzhang commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2150957887
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.io
+
+import java.io.InputStream
+
+import scala.collection.Seq
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.compress._
+
+import org.apache.spark.{SparkConf, SparkEnv}
+import org.apache.spark.internal.config
+import org.apache.spark.io.{CompressionCodec => SparkCompressionCodec}
+
+object HadoopCodecStreams {
Review Comment:
scaladoc the functions in here?
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2982960085
> There are users who have files with either non-standard extensions or no extensions at all.
@sandip-db to handle "non-standard extensions", you just need to register another Hadoop codec, for example
```
new org.apache.hadoop.io.compress.GzipCodec {
override def getDefaultExtension: String = "gzip"
}
```
for "no extensions" compressed text files, I'm not sure if this is a valid use case(see my last comment)
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on PR #51182:
URL: https://github.com/apache/spark/pull/51182#issuecomment-2984471288
> also, please be careful with that Hadoop codec may have different behavior with Spark/Unix tool codec, for example, HADOOP-12990(lz4)
Thanks for bringing this to my attention. We are not adding support for ZSTD compression using Spark codec yet. This PR just adds decompression support.
@cloud-fan @pan3793 Do you have any further concern with this PR. Can this be merged?
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2153373392
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -3686,6 +3687,96 @@ abstract class CSVSuite
parameters = Map("columnName" -> "`v`", "columnType" -> "\"VARIANT\"", "format" -> "CSV")
)
}
+
+ private def createTestFiles(dir: File, fileFormatWriter: Boolean,
Review Comment:
shall we put the test in `FileBasedDataSourceSuite`, so that it's easier to share code and test different formats?
--
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
Re: [PR] [SPARK-52482][SQL][CORE] ZStandard support for file data source reader [spark]
Posted by "sandip-db (via GitHub)" <gi...@apache.org>.
sandip-db commented on code in PR #51182:
URL: https://github.com/apache/spark/pull/51182#discussion_r2153381034
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -3686,6 +3687,96 @@ abstract class CSVSuite
parameters = Map("columnName" -> "`v`", "columnType" -> "\"VARIANT\"", "format" -> "CSV")
)
}
+
+ private def createTestFiles(dir: File, fileFormatWriter: Boolean,
Review Comment:
I thought about this. However, each format has some special cases that requires generating the files slightly differently:
- TextFileFormat has wholeText option that reads the whole file
- CSV has header option
- Json has single and multiline options
--
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