Skip to content

[SPARK-21839][SQL] Support SQL config for ORC compression #19055

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dongjoon-hyun wants to merge 10 commits into from
Closed

[SPARK-21839][SQL] Support SQL config for ORC compression #19055

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
f3ccfec
[SPARK-21839][SQL] Support SQL config for ORC compression
dongjoon-hyun Aug 25, 2017
5998c29
fix doc.
dongjoon-hyun Aug 25, 2017
afbb6f2
address comments.
dongjoon-hyun Aug 28, 2017
437d181
add lzo.
dongjoon-hyun Aug 29, 2017
aafbfbb
Address comments.
dongjoon-hyun Aug 29, 2017
8aebcd3
fix
dongjoon-hyun Aug 29, 2017
94e624e
remove unused import.
dongjoon-hyun Aug 30, 2017
34e2845
update comments according to the review comment.
dongjoon-hyun Aug 30, 2017
4c0300c
Update comments in readwriter.py, too.
dongjoon-hyun Aug 30, 2017
9620e46
fix
dongjoon-hyun Aug 30, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,6 +322,15 @@ object SQLConf {
.booleanConf
.createWithDefault(true)

// We can add LZO after Apache ORC 1.4.0 is used.
val ORC_COMPRESSION = buildConf("spark.sql.orc.compression.codec")
.doc("Sets the compression codec use when writing ORC files. Acceptable values include: " +
"uncompressed, snappy, zlib.")
.stringConf
.transform(_.toLowerCase(Locale.ROOT))
.checkValues(Set("none", "uncompressed", "snappy", "zlib"))

@HyukjinKwon HyukjinKwon Aug 29, 2017 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dongjoon-hyun, I think my only main concern is inconsistency between compression option and this config. If lzo is an unknown key and it directly throws an exception (or even this was not there in the first place), I would have been okay but it looks attempting to find org.apache.hadoop.hive.ql.io.orc.LzoCodec:

java.lang.IllegalArgumentException: LZO is not available.
	at org.apache.hadoop.hive.ql.io.orc.WriterImpl.createCodec(WriterImpl.java:331)
	at org.apache.hadoop.hive.ql.io.orc.WriterImpl.<init>(WriterImpl.java:201)
	at org.apache.hadoop.hive.ql.io.orc.OrcFile.createWriter(OrcFile.java:464)
	at org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat$OrcRecordWriter.write(OrcOutputFormat.java:74)
	at org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat$OrcRecordWriter.write(OrcOutputFormat.java:55)
       ...

...
      case LZO:
        try {
          Class<? extends CompressionCodec> lzo =
              (Class<? extends CompressionCodec>)
                  JavaUtils.loadClass("org.apache.hadoop.hive.ql.io.orc.LzoCodec");
          return lzo.newInstance();
        } catch (ClassNotFoundException e) {
          throw new IllegalArgumentException("LZO is not available.", e);
        } catch (InstantiationException e) {
          throw new IllegalArgumentException("Problem initializing LZO", e);
        } catch (IllegalAccessException e) {
          throw new IllegalArgumentException("Insufficient access to LZO", e);
        }
...

This appears that if we provide org.apache.hadoop.hive.ql.io.orc.LzoCodec anyhow in the classpath (as an extream case, the one implemented by an user), it should work, if I read this correctly. This should be almost seldom though.

Does your set of ORC related PRs eventually support lzo correctly?

@dongjoon-hyun dongjoon-hyun Aug 29, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see. You mean user-provided hive or libraries. It makes sense. I will add lzo too here. Thank you for pointing out that.

.createWithDefault("snappy")

val ORC_FILTER_PUSHDOWN_ENABLED = buildConf("spark.sql.orc.filterPushdown")
.doc("When true, enable filter pushdown for ORC files.")
.booleanConf
Expand Down Expand Up @@ -998,6 +1007,8 @@ class SQLConf extends Serializable with Logging {

def useCompression: Boolean = getConf(COMPRESS_CACHED)

def orcCompressionCodec: String = getConf(ORC_COMPRESSION)

def parquetCompressionCodec: String = getConf(PARQUET_COMPRESSION)

def parquetCacheMetadata: Boolean = getConf(PARQUET_CACHE_METADATA)
Expand Down
2 changes: 1 addition & 1 deletion sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ class OrcFileFormat extends FileFormat with DataSourceRegister with Serializable
job: Job,
options: Map[String, String],
dataSchema: StructType): OutputWriterFactory = {
val orcOptions = new OrcOptions(options)
val orcOptions = new OrcOptions(options, sparkSession.sessionState.conf)

val configuration = job.getConfiguration

Expand Down
17 changes: 10 additions & 7 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcOptions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,30 +20,33 @@ package org.apache.spark.sql.hive.orc
import java.util.Locale

import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.internal.SQLConf

/**
* Options for the ORC data source.
*/
private[orc] class OrcOptions(@transient private val parameters: CaseInsensitiveMap[String])
private[orc] class OrcOptions(
@transient private val parameters: CaseInsensitiveMap[String],
@transient private val sqlConf: SQLConf)
extends Serializable {

import OrcOptions._

def this(parameters: Map[String, String]) = this(CaseInsensitiveMap(parameters))
def this(parameters: Map[String, String], sqlConf: SQLConf) =
this(CaseInsensitiveMap(parameters), sqlConf)

/**
* Compression codec to use. By default snappy compression.
* Compression codec to use.
* Acceptable values are defined in [[shortOrcCompressionCodecNames]].
*/
val compressionCodec: String = {
// `orc.compress` is a ORC configuration. So, here we respect this as an option but
// `compression` has higher precedence than `orc.compress`. It means if both are set,
// we will use `compression`.
// `compression`, `orc.compress`, and `spark.sql.orc.compression.codec` is used in order.

@gatorsmile gatorsmile Aug 30, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is used in order -> are in order of precedence from highest to lowest

val orcCompressionConf = parameters.get(OrcRelation.ORC_COMPRESSION)
val codecName = parameters
.get("compression")
.orElse(orcCompressionConf)
.getOrElse("snappy").toLowerCase(Locale.ROOT)
.getOrElse(sqlConf.orcCompressionCodec)

@HyukjinKwon HyukjinKwon Aug 30, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we update the default values for consistency with Parquet one? :

spark/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

Line 520 in 3c0c2d0

* <li>`compression` (default `snappy`): compression codec to use when saving to file. This can be

spark/python/pyspark/sql/readwriter.py

Line 855 in 51620e2

default value, ``snappy``.

@dongjoon-hyun dongjoon-hyun Aug 30, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default value is snappy, isn't it?

.createWithDefault("snappy")

@HyukjinKwon HyukjinKwon Aug 30, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was thinking like:

spark/python/pyspark/sql/readwriter.py

Lines 751 to 753 in 51620e2

This will override ``spark.sql.parquet.compression.codec``. If None
is set, it uses the value specified in
``spark.sql.parquet.compression.codec``.

Wouldn't we use the value set in spark.sql.parquet.compression.codec by default if compression is unset via option API?

@HyukjinKwon HyukjinKwon Aug 30, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, I thought the purpose of this configuration is rather for setting the default compression codec for ORC datasource ..

@dongjoon-hyun dongjoon-hyun Aug 30, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. This is the priority. If compression and orc.compression is unset via option, we use SQLConf.
compression -> orc.compression -> spark.sql.orc.compression.codec

@dongjoon-hyun dongjoon-hyun Aug 30, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The main purpose of this PR is to support users to control ORC compression by using SQLConf, too.

@dongjoon-hyun dongjoon-hyun Aug 30, 2017 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default codec is unchanged and the priority is the same. Also, all previous user-given options are respected.

@HyukjinKwon HyukjinKwon Aug 30, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, it looks I had to be clear. I meant fixing the comment for default value from snappy to spark.sql.orc.compression.codec.

@dongjoon-hyun dongjoon-hyun Aug 30, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you! I'll fix that.

.toLowerCase(Locale.ROOT)
if (!shortOrcCompressionCodecNames.contains(codecName)) {
val availableCodecs = shortOrcCompressionCodecNames.keys.map(_.toLowerCase(Locale.ROOT))
throw new IllegalArgumentException(s"Codec [$codecName] " +
Expand Down
24 changes: 22 additions & 2 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcSourceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,13 @@
package org.apache.spark.sql.hive.orc

import java.io.File
import java.util.Locale

@HyukjinKwon HyukjinKwon Aug 30, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks unused in this test.

@dongjoon-hyun dongjoon-hyun Aug 30, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oops. Thanks!


import org.scalatest.BeforeAndAfterAll

import org.apache.spark.sql.{QueryTest, Row}
import org.apache.spark.sql.hive.HiveExternalCatalog
import org.apache.spark.sql.hive.test.TestHiveSingleton
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -149,7 +150,8 @@ abstract class OrcSuite extends QueryTest with TestHiveSingleton with BeforeAndA
}

test("SPARK-18433: Improve DataSource option keys to be more case-insensitive") {
assert(new OrcOptions(Map("Orc.Compress" -> "NONE")).compressionCodec == "NONE")
val conf = sqlContext.sessionState.conf
assert(new OrcOptions(Map("Orc.Compress" -> "NONE"), conf).compressionCodec == "NONE")
}

test("SPARK-19459/SPARK-18220: read char/varchar column written by Hive") {
Expand Down Expand Up @@ -194,6 +196,24 @@ abstract class OrcSuite extends QueryTest with TestHiveSingleton with BeforeAndA
Utils.deleteRecursively(location)
}
}

test("SPARK-21839: Add SQL config for ORC compression") {
val conf = sqlContext.sessionState.conf
assert(new OrcOptions(Map.empty[String, String], conf).compressionCodec == "SNAPPY")

@gatorsmile gatorsmile Aug 29, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add a comment here to explain the test scenario.
// to test the default of spark.sql.orc.compression.codec is snappy

@dongjoon-hyun dongjoon-hyun Aug 29, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure


// OrcOptions's parameters have a higher priority than SQL configuration.
withSQLConf(SQLConf.ORC_COMPRESSION.key -> "uncompressed") {
assert(new OrcOptions(Map.empty[String, String], conf).compressionCodec == "NONE")
assert(
new OrcOptions(Map("orc.compress" -> "zlib"), conf).compressionCodec == "ZLIB")

@gatorsmile gatorsmile Aug 29, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also please add another scenario when users specify compression

}

Seq("NONE", "SNAPPY", "ZLIB").foreach { c =>
withSQLConf(SQLConf.ORC_COMPRESSION.key -> c) {
assert(new OrcOptions(Map.empty[String, String], conf).compressionCodec == c)
}
}
}
}

class OrcSourceSuite extends OrcSuite {
Expand Down