You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by ksakellis <gi...@git.apache.org> on 2014/11/05 22:33:43 UTC
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
GitHub user ksakellis opened a pull request:
https://github.com/apache/spark/pull/3119
[SPARK-4079] [CORE] Default to LZF if Snappy not available
By default, snappy is the compression codec used. If Snappy is not available, Spark currently throws a stack trace. Now Spark falls back to LZF if Snappy is not available on the cluster and logs a warning message.
The only exception is if the user has explicitly set spark.io.compression.codec=snappy. In this case, if snappy is not available, an IllegalArgumentException is thrown.
Because of the way the Snappy library uses static initialization, it was very difficult in a unit test to simulate Snappy not being available. The only way I could think of was to create multiple classloaders which seemed excessive. As a result, most of this was tested adhoc on a test cluster by modifying the system property: org.xerial.snappy.use.systemlib=true which caused Snappy to not load and thus triggering this logic.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ksakellis/spark kostas-spark-4079
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/3119.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3119
----
commit c8bc9db38f461ed4652ebf1ec70ef967b0d8f040
Author: Kostas Sakellis <ko...@cloudera.com>
Date: 2014-11-05T02:26:12Z
[SPARK-4079] [CORE] Default to LZF if Snappy not available
By default, snappy is the compression codec used.
If Snappy is not available, Spark currently throws
a stack trace. Now Spark falls back to LZF
if Snappy is not available on the cluster and logs
a warning message.
The only exception is if the user has explicitly
set spark.io.compression.codec=snappy. In this
case, if snappy is not available, an
IllegalArgumentException is thrown.
Because of the way the Snappy library uses static
initialization, it was very difficult in a unit test to
simulate Snappy not being available. The only way I
could think of was to create multiple classloaders
which seemed excessive. As a result, most of this was tested
adhoc on a test cluster by modifying the system property:
org.xerial.snappy.use.systemlib=true which caused Snappy
to not load and thus triggering this logic.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r20829821
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,27 +43,37 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
--- End diff --
This trait is exposed to adding a method here will actually break binary compatibility with any user-defined compression codecs. We've usually tried not to do this in the past unless we had a situation where there was no work around, so what about just dealing with this directly inside the Snappy codec? I.e. just check when the snappy codec is instantiated and throw this exception if it's not available. It's slightly less general, but a simpler patch and we don't risk breaking user-defined codecs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-63399499
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23505/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-63399491
[Test build #23505 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/23505/consoleFull) for PR 3119 at commit [`3847eef`](https://github.com/apache/spark/commit/3847eef503f4d1e5494c740efe9ab9364bc8d78b).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r19908142
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,28 +43,52 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
+}
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
+ private val configKey = "spark.io.compression.codec"
private val shortCompressionCodecNames = Map(
"lz4" -> classOf[LZ4CompressionCodec].getName,
"lzf" -> classOf[LZFCompressionCodec].getName,
"snappy" -> classOf[SnappyCompressionCodec].getName)
def createCodec(conf: SparkConf): CompressionCodec = {
- createCodec(conf, conf.get("spark.io.compression.codec", DEFAULT_COMPRESSION_CODEC))
+ conf.getOption(configKey)
+ .map(createCodec(conf, _))
+ .orElse(createCodecFromName(conf, DEFAULT_COMPRESSION_CODEC))
+ .orElse({
+ logWarning("Default codec " + DEFAULT_COMPRESSION_CODEC +
+ " is unavailable. Faling back to " + FALLBACK_COMPRESSION_CODEC)
+ createCodecFromName(conf, FALLBACK_COMPRESSION_CODEC)
+ })
+ .getOrElse(throw new IllegalArgumentException("The codec [" +
--- End diff --
I think this message will be a little confusing, since only the fallback codec would be listed and it might be easy to miss the warning log. Maybe rephrase it as "No default compression codec implementation is available"?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by ksakellis <gi...@git.apache.org>.
Github user ksakellis commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62447986
@vanzin regarding testing... yes, I did try what you recommended - I added a note in the commit message. Snappy does some static initialization so if it was unavailable when you first referenced the class, it will never be available again. So i can create a unit test but then all the other snappy unit tests fail. This is why i mentioned that I would need multiple classloaders to get around this issue.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by ksakellis <gi...@git.apache.org>.
Github user ksakellis commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r21111000
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,27 +43,37 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
--- End diff --
oh geez. thanks for the link. it makes sense but yes, not what you would expect from traits. I'll update the pr.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-63388094
Jenkins, test this please. Hm... they should be retriggering automatically.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-63388314
[Test build #23505 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/23505/consoleFull) for PR 3119 at commit [`3847eef`](https://github.com/apache/spark/commit/3847eef503f4d1e5494c740efe9ab9364bc8d78b).
* This patch merges cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r22190260
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -120,6 +129,12 @@ class LZFCompressionCodec(conf: SparkConf) extends CompressionCodec {
@DeveloperApi
class SnappyCompressionCodec(conf: SparkConf) extends CompressionCodec {
+ try {
+ Snappy.getNativeLibraryVersion
--- End diff --
I was thinking about this, but it's private: https://github.com/xerial/snappy-java/blob/master/src/main/java/org/xerial/snappy/SnappyLoader.java#L83
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-67082812
Current approach LGTM. Had just one small question.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62313527
Could this instead just throw an exception when Snappy is configured but not supported? We typically try not to silently mutate configs in the background in favor of giving users an actionable exception. I think this could be accomplished by just modifying `SnappyCompressionCodec` to guard the creation of an input stream or output stream with a check as to whether Snappy is enabled, and throw an exception if it is not enabled.
The current approach could lead to very confusing failure behavior. For instance say a user has the Snappy native library installed on some machines but not others. What will happen is that there will be a stream corruption exception somewhere inside of Spark where one node writes data as Snappy and another reads it as LZF. And to figure out what caused this a user will have to troll through executor logs for a somewhat innocuous looking `WARN` statement.
@rxin designed this codec interface (I think) so maybe he has more comments also.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r19907994
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,28 +43,52 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
+}
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
+ private val configKey = "spark.io.compression.codec"
private val shortCompressionCodecNames = Map(
"lz4" -> classOf[LZ4CompressionCodec].getName,
"lzf" -> classOf[LZFCompressionCodec].getName,
"snappy" -> classOf[SnappyCompressionCodec].getName)
def createCodec(conf: SparkConf): CompressionCodec = {
- createCodec(conf, conf.get("spark.io.compression.codec", DEFAULT_COMPRESSION_CODEC))
+ conf.getOption(configKey)
+ .map(createCodec(conf, _))
--- End diff --
nit: indented too far (Spark's convention is 2 spaces for the continuation line).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r19908270
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,28 +43,52 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
+}
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
+ private val configKey = "spark.io.compression.codec"
private val shortCompressionCodecNames = Map(
"lz4" -> classOf[LZ4CompressionCodec].getName,
"lzf" -> classOf[LZFCompressionCodec].getName,
"snappy" -> classOf[SnappyCompressionCodec].getName)
def createCodec(conf: SparkConf): CompressionCodec = {
- createCodec(conf, conf.get("spark.io.compression.codec", DEFAULT_COMPRESSION_CODEC))
+ conf.getOption(configKey)
+ .map(createCodec(conf, _))
+ .orElse(createCodecFromName(conf, DEFAULT_COMPRESSION_CODEC))
+ .orElse({
+ logWarning("Default codec " + DEFAULT_COMPRESSION_CODEC +
+ " is unavailable. Faling back to " + FALLBACK_COMPRESSION_CODEC)
+ createCodecFromName(conf, FALLBACK_COMPRESSION_CODEC)
+ })
+ .getOrElse(throw new IllegalArgumentException("The codec [" +
+ FALLBACK_COMPRESSION_CODEC + "] is not available."))
}
def createCodec(conf: SparkConf, codecName: String): CompressionCodec = {
+ createCodecFromName(conf, codecName)
+ .getOrElse(throw new IllegalArgumentException("The specified codec [" +
+ codecName + "] is not available."))
+ }
+
+ private def createCodecFromName(conf: SparkConf, codecName : String)
+ : Option[CompressionCodec] = {
val codecClass = shortCompressionCodecNames.getOrElse(codecName.toLowerCase, codecName)
- val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
- .getConstructor(classOf[SparkConf])
- ctor.newInstance(conf).asInstanceOf[CompressionCodec]
+ try {
+ val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
+ .getConstructor(classOf[SparkConf])
+ Some(ctor.newInstance(conf).asInstanceOf[CompressionCodec])
+ .filter(_.isAvailable())
--- End diff --
nit: fits in previous line? (limit is 100 chars)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by ksakellis <gi...@git.apache.org>.
Github user ksakellis commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r21922602
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -120,6 +129,12 @@ class LZFCompressionCodec(conf: SparkConf) extends CompressionCodec {
@DeveloperApi
class SnappyCompressionCodec(conf: SparkConf) extends CompressionCodec {
+ try {
+ Snappy.getNativeLibraryVersion
--- End diff --
@pwendell can you point to the api you are talking about?
https://github.com/xerial/snappy-java/tree/1.1.1.6/src/main/java/org/xerial/snappy
I don't see a Snappy.isLoaded() API. There is a member variable in SnappyLoader but that is private.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r19909185
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,28 +43,52 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
+}
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
+ private val configKey = "spark.io.compression.codec"
private val shortCompressionCodecNames = Map(
"lz4" -> classOf[LZ4CompressionCodec].getName,
"lzf" -> classOf[LZFCompressionCodec].getName,
"snappy" -> classOf[SnappyCompressionCodec].getName)
def createCodec(conf: SparkConf): CompressionCodec = {
- createCodec(conf, conf.get("spark.io.compression.codec", DEFAULT_COMPRESSION_CODEC))
+ conf.getOption(configKey)
+ .map(createCodec(conf, _))
+ .orElse(createCodecFromName(conf, DEFAULT_COMPRESSION_CODEC))
--- End diff --
So, I know I suggested this, but I think this doesn't do the same thing as the PR description says.
If the config option exists and the requested codec is not available, this will always try the default codecs. Instead, this should probably be something like:
if (conf.contains(configKey)) {
createCodec(...).getOrElse(throw new Blah("requested codec cannot be loaded"))
} else
createCodec(DEFAULT).orElse(createCodec(FALLBACK)).getOrElse(throw new Blah("cannot load defaults"))
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r20835119
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,27 +43,37 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
--- End diff --
By the way - the fact that you cannot safely add a method with a default implementation to a trait is not intuitive at all. You can checkout this article for more information why this is:
http://stackoverflow.com/questions/18366817/is-adding-a-trait-method-with-implementation-breaking-backward-compatibility
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-61886283
Just small nits. re: testing, can't you write a test where you set the system property to make snappy fail as part of that test?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r19908213
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,28 +43,52 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
+}
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
+ private val configKey = "spark.io.compression.codec"
private val shortCompressionCodecNames = Map(
"lz4" -> classOf[LZ4CompressionCodec].getName,
"lzf" -> classOf[LZFCompressionCodec].getName,
"snappy" -> classOf[SnappyCompressionCodec].getName)
def createCodec(conf: SparkConf): CompressionCodec = {
- createCodec(conf, conf.get("spark.io.compression.codec", DEFAULT_COMPRESSION_CODEC))
+ conf.getOption(configKey)
+ .map(createCodec(conf, _))
+ .orElse(createCodecFromName(conf, DEFAULT_COMPRESSION_CODEC))
+ .orElse({
+ logWarning("Default codec " + DEFAULT_COMPRESSION_CODEC +
+ " is unavailable. Faling back to " + FALLBACK_COMPRESSION_CODEC)
+ createCodecFromName(conf, FALLBACK_COMPRESSION_CODEC)
+ })
+ .getOrElse(throw new IllegalArgumentException("The codec [" +
+ FALLBACK_COMPRESSION_CODEC + "] is not available."))
}
def createCodec(conf: SparkConf, codecName: String): CompressionCodec = {
+ createCodecFromName(conf, codecName)
+ .getOrElse(throw new IllegalArgumentException("The specified codec [" +
--- End diff --
Interpolation is probably prettier here:
s"The specified codec [$codecName] is not available."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by ksakellis <gi...@git.apache.org>.
Github user ksakellis commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-64050068
@pwendell Can you please review? Is there anything else we need before we can merge?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r20473829
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -58,9 +60,15 @@ private[spark] object CompressionCodec {
def createCodec(conf: SparkConf, codecName: String): CompressionCodec = {
val codecClass = shortCompressionCodecNames.getOrElse(codecName.toLowerCase, codecName)
- val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
- .getConstructor(classOf[SparkConf])
- ctor.newInstance(conf).asInstanceOf[CompressionCodec]
+ val codec = try {
+ val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
+ .getConstructor(classOf[SparkConf])
+ Some(ctor.newInstance(conf).asInstanceOf[CompressionCodec])
+ } catch {
+ case e: ClassNotFoundException => None
+ }
+ codec.filter(_.isAvailable())
+ .getOrElse(throw new IllegalArgumentException(s"Codec [$codecName] is not available."))
--- End diff --
This seems like a good idea.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62464530
BTW with sort based shuffle, it might be ok to roll back the default compression to lzf, since we no longer open a large number of concurrent streams. It is now only one per map task.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-61885458
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r21864897
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -120,6 +129,12 @@ class LZFCompressionCodec(conf: SparkConf) extends CompressionCodec {
@DeveloperApi
class SnappyCompressionCodec(conf: SparkConf) extends CompressionCodec {
+ try {
+ Snappy.getNativeLibraryVersion
--- End diff --
Should we instead check for `Snappy.isLoaded()`? This way seems more roundabout.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-64280161
It's not blocking 1.2 so I have other things on the plate. Should be able to get to it soon.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by ksakellis <gi...@git.apache.org>.
Github user ksakellis commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62995656
@pwendell Can you please trigger the jenkins tests for this pr?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/3119
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r20331626
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -42,10 +43,11 @@ trait CompressionCodec {
def compressedOutputStream(s: OutputStream): OutputStream
def compressedInputStream(s: InputStream): InputStream
-}
+ def isAvailable() : Boolean = true
+}
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
--- End diff --
`extends Logging` is not needed anymore?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62348827
I agree with @pwendell - it'd be better to just fail fast here rather than picking LZF based on the the availability of the library.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by ksakellis <gi...@git.apache.org>.
Github user ksakellis commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62485529
@pwendell good point. I'll just make this throw an exception when the you try to create a SnappyCompressionCodec and snappy is not available. I think that is better than waiting until you create an input or output stream - fails faster.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Default to LZF if Snappy n...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3119#issuecomment-62226594
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-4079] [CORE] Consolidates Errors if a C...
Posted by vanzin <gi...@git.apache.org>.
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/3119#discussion_r20331903
--- Diff: core/src/main/scala/org/apache/spark/io/CompressionCodec.scala ---
@@ -58,9 +60,15 @@ private[spark] object CompressionCodec {
def createCodec(conf: SparkConf, codecName: String): CompressionCodec = {
val codecClass = shortCompressionCodecNames.getOrElse(codecName.toLowerCase, codecName)
- val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
- .getConstructor(classOf[SparkConf])
- ctor.newInstance(conf).asInstanceOf[CompressionCodec]
+ val codec = try {
+ val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
+ .getConstructor(classOf[SparkConf])
+ Some(ctor.newInstance(conf).asInstanceOf[CompressionCodec])
+ } catch {
+ case e: ClassNotFoundException => None
+ }
+ codec.filter(_.isAvailable())
+ .getOrElse(throw new IllegalArgumentException(s"Codec [$codecName] is not available."))
--- End diff --
Hmm. This is fine, but one suggestion in the bug was to have extra wording for the case where the default codec is not available, suggesting to the user to change the configuration to set the codec to LZF instead. I think that would be a nice thing to have, since that's probably the most common case of this exception happening.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org