[SPARK-19525][CORE] Compressing checkpoints. #17024

aramesh117 · 2017-02-22T05:24:48Z

Spark Streaming's latency performance improves greatly for smaller batches if we enable compression of
checkpoints.

What changes were proposed in this pull request?

Compress each partition before writing to persistent file system.
Decompress each partition before reading from persistent file system.
Default behavior should be to not compress.
Add logging for checkpoint durations for A/B testing with and without compression enabled.

How was this patch tested?

This was tested using existing unit tests for backwards compatibility and with new tests for this functionality. It has also been used in our production system for almost a year.

Please review http://spark.apache.org/contributing.html before opening a pull request.

Spark's performance improves greatly if we enable compression of checkpoints.

AmplabJenkins · 2017-02-22T05:27:15Z

Can one of the admins verify this patch?

mridulm

This would be a great addition to checkpoint'ing, thanks for contributing !
I have left some comments.

mridulm · 2017-02-22T09:42:09Z

core/src/main/scala/org/apache/spark/io/CompressionCodec.scala

@@ -95,6 +95,7 @@ private[spark] object CompressionCodec {
  val FALLBACK_COMPRESSION_CODEC = "snappy"
  val DEFAULT_COMPRESSION_CODEC = "lz4"
  val ALL_COMPRESSION_CODECS = shortCompressionCodecNames.values.toSeq
+  val ALL_COMPRESSION_CODECS_SHORT: Set[String] = shortCompressionCodecNames.keySet


Instead of exposing this and supporting only short codec names for checkpoint, the pattern should be same as in rest of spark code when dealing with codec's.

sparkConf.getOption("spark.checkpoint.compress.codec").map(c => logInfo(s"Compressing checkpoint using $c.") CompressionCodec.createCodec(conf, c). compressedInputStream or compressedOutputStream ).getOrElse(fileStream)

This will ensure that support for checkpoint compression is in line with rest of spark (short and long classes, no need to introduce 'none')

Note: you will need to change fileStream to a lazy val - so that if codec creation throws exception, we dont leave dangling streams around (with limited block visibility scope to fileStream)

mridulm · 2017-02-22T09:47:54Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

@@ -133,9 +134,13 @@ private[spark] object ReliableCheckpointRDD extends Logging {
    val broadcastedConf = sc.broadcast(
      new SerializableConfiguration(sc.hadoopConfiguration))
    // TODO: This is expensive because it computes the RDD again unnecessarily (SPARK-8582)
+    logInfo(s"The checkpoint compression codec is " +


This should be logged if compression is enabled (none is not a supported compression codec).
It could also be rolled into the timing info log message below.

mridulm · 2017-02-22T09:48:46Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

    } else {
      // This is mainly for testing purpose
      fs.create(tempOutputPath, false, bufferSize,
        fs.getDefaultReplication(fs.getWorkingDirectory), blockSize)
    }
    val serializer = env.serializer.newInstance()
    val serializeStream = serializer.serializeStream(fileOutputStream)
+    logInfo(s"Starting to write to checkpoint file $tempOutputPath.")


This will make the logs verbose.
If it does help with debugging, you could make it logTrace - or remove it entirely.

My thought was that since checkpointing shouldn't be done too frequently anyway, this won't make the logs too verbose in the executor, and may be helpful for debugging after issues with checkpointing have already occurred. I'll make it logDebug for now, is this okay?

logDebug should be fine too.

mridulm · 2017-02-22T09:49:10Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

@@ -197,6 +212,7 @@ private[spark] object ReliableCheckpointRDD extends Logging {
        }
      }
    }
+    logInfo(s"Checkpointing took ${System.currentTimeMillis() - startTimeMs} ms.")


Add codec (if used) here.

mridulm · 2017-02-22T09:49:50Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

+        logInfo(s"Compressing using $checkpointCodec.")
+        compressionCodec.compressedOutputStream(fileStream)
+      } else {
+        fileStream


This repeated pattern can be rewritten as indicated above https://github.com/apache/spark/pull/17024/files#r102418860

mridulm · 2017-02-22T09:50:03Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

+        CompressionCodec.createCodec(env.conf, checkpointCodec).compressedInputStream(fileStream)
+      } else {
+        fileStream
+      }


Use https://github.com/apache/spark/pull/17024/files#r102418860

aramesh117 · 2017-02-22T19:47:54Z

@mridulm Thank you so much! I will definitely update with your suggestions.

mridulm · 2017-02-23T00:08:16Z

+CC @tdas

aramesh117 · 2017-02-25T23:10:59Z

@mridulm I've added a new commit. Thank you for the review! :)

mridulm · 2017-02-27T00:15:49Z

@aramesh117 looks good !
I would also like someone working on streaming to chime in - since that is a common usecase for checkpoint.

+CC @tdas, @zsxwing

mridulm · 2017-02-27T00:16:31Z

I wonder if adding an extension (to the file) helps based on codec ...

aramesh117 · 2017-03-07T21:32:17Z

@mridulm Sure I can add in a file extension based on the codec being used. But is there a specific use case that adding an extension would solve?

mridulm · 2017-03-07T21:47:57Z

It makes it possible to identify what the data within the file is (compressed or not) - for user's perusal (it does not change anything for the application, that is true).
But before you change code, I would like comments from streaming developers on this change.

aramesh117 · 2017-04-12T08:14:07Z

@mridulm Waiting for @tdas and @zsxwing has taken more than a month now. Is there any other way we can reach them or is there anyone else that can take a look at this merge request? This is a critical change that is needed for Conviva's use case if we are to upgrade to later versions of Spark.

mridulm · 2017-04-12T09:29:31Z

@aramesh117 Unfortunately, since this heavily affects streaming, I cannot sign off on it without someone more familiar with spark streaming reviews it as well.

zsxwing

Sorry for the delay. Made one pass.

zsxwing · 2017-04-17T21:44:57Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

 import org.apache.spark.util.{SerializableConfiguration, Utils}

+


nit: please remove unnecessary space changes

zsxwing · 2017-04-17T21:50:08Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

    sc.runJob(originalRDD,
      writePartitionToCheckpointFile[T](checkpointDirPath.toString, broadcastedConf) _)

+    logInfo(s"Checkpointing took ${System.currentTimeMillis() - startTime} ms.")
+    sc.conf.getOption("spark.checkpoint.compress.codec").foreach(codec => {


For consistency, I suggest we just add a new config spark.checkpoint.compress which means whether to enable checkpoint compression. See

spark/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

Line 74 in b56ad2b

compressionCodec = if (conf.getBoolean("spark.broadcast.compress", true)) {

for example.

zsxwing · 2017-04-17T21:51:32Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

@@ -133,9 +136,14 @@ private[spark] object ReliableCheckpointRDD extends Logging {
    val broadcastedConf = sc.broadcast(
      new SerializableConfiguration(sc.hadoopConfiguration))
    // TODO: This is expensive because it computes the RDD again unnecessarily (SPARK-8582)
+    val startTime = System.currentTimeMillis()


nit: please use nanoTime to measure the duration. See https://github.com/databricks/scala-style-guide/tree/f6cce9ab32e7b288638f2f1615d20a3b6d16ef2e#misc_currentTimeMillis_vs_nanoTime

zsxwing · 2017-04-17T21:52:32Z

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

    } else {
      // This is mainly for testing purpose
      fs.create(tempOutputPath, false, bufferSize,
        fs.getDefaultReplication(fs.getWorkingDirectory), blockSize)
    }
    val serializer = env.serializer.newInstance()
    val serializeStream = serializer.serializeStream(fileOutputStream)
+    logTrace(s"Starting to write to checkpoint file $tempOutputPath.")
+    val startTimeMs = System.currentTimeMillis()


same as above

zsxwing · 2017-04-17T21:52:46Z

core/src/test/scala/org/apache/spark/CheckpointSuite.scala

 import org.apache.spark.rdd._
 import org.apache.spark.storage.{BlockId, StorageLevel, TestBlockId}
 import org.apache.spark.util.Utils

+


nit: please move unnecessary changes.

zsxwing · 2017-04-17T21:52:59Z

core/src/test/scala/org/apache/spark/CheckpointSuite.scala

    sc = new SparkContext("local", "test")
    sc.setCheckpointDir(checkpointDir.toString)
  }

+


nit: please move unnecessary changes.

zsxwing · 2017-04-17T22:03:14Z

core/src/test/scala/org/apache/spark/CheckpointSuite.scala

+    testBasicCheckpoint(sc, reliableCheckpoint)
+  }
+
+  runTest("compression with snappy", skipLocalCheckpoint = true) { _: Boolean =>


After you change the config to spark.checkpoint.compress, you don't need to test all compression codecs. Just write one test for the default codec. Others should be covered in CompressionCodecSuite.

For the new test, I think we just need one simple test. And if we put it into a new suite (e.g., the below example), then we don't need to touch the existing codes.

class CheckpointCompressionSuite extends SparkFunSuite with LocalSparkContext { test("checkpoint compression") { val checkpointDir = File.createTempFile("temp", "", Utils.createTempDir()) try { val conf = new SparkConf().set("spark.checkpoint.compress", "true") sc = new SparkContext("local", "test", conf) sc.setCheckpointDir(checkpointDir.toString) val rdd = sc.makeRDD(1 to 20, numSlices = 1) rdd.checkpoint() assert(rdd.collect().toSeq === (1 to 20)) val checkpointPath = new Path(rdd.getCheckpointFile.get) val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration) val checkpointFile = fs.listStatus(checkpointPath).map(_.getPath).find(_.getName.startsWith("part-")).get // Verify the checkpoint file can be decompressed val compressedInputStream = CompressionCodec.createCodec(conf) .compressedInputStream(fs.open(checkpointFile)) ByteStreams.toByteArray(compressedInputStream) // Verify that the compressed content can be read back assert(rdd.collect().toSeq === (1 to 20)) } finally { Utils.deleteRecursively(checkpointDir) } } }

zsxwing · 2017-04-17T22:14:23Z

core/src/test/scala/org/apache/spark/CheckpointSuite.scala

@@ -238,6 +241,42 @@ trait RDDCheckpointTester { self: SparkFunSuite =>
  protected def generateFatPairRDD(): RDD[(Int, Int)] = {
    new FatPairRDD(sparkContext.makeRDD(1 to 100, 4), partitioner).mapValues(x => x)
  }
+
+  protected def testBasicCheckpoint(sc: SparkContext, reliableCheckpoint: Boolean): Unit = {


nit: does this one test any special logic? If it's covered by other tests, not need to add it to increase the test time.

zsxwing · 2017-04-26T17:30:26Z

@aramesh117 do you have time to work on this PR recently? We need to merge this PR ASAP in order to get it into 2.2.0. Thanks!

zsxwing · 2017-04-27T20:49:40Z

@aramesh117 I just opened #17789 to finish the rest work. All credits will go to you when merging the new PR.

## What changes were proposed in this pull request? This PR adds RDD checkpoint compression support and add a new config `spark.checkpoint.compress` to enable/disable it. Credit goes to aramesh117 Closes #17024 ## How was this patch tested? The new unit test. Author: Shixiong Zhu <[email protected]> Author: Aaditya Ramesh <[email protected]> Closes #17789 from zsxwing/pr17024. (cherry picked from commit 77bcd77) Signed-off-by: Shixiong Zhu <[email protected]>

This PR adds RDD checkpoint compression support and add a new config `spark.checkpoint.compress` to enable/disable it. Credit goes to aramesh117 Closes apache#17024 The new unit test. Author: Shixiong Zhu <[email protected]> Author: Aaditya Ramesh <[email protected]> Closes apache#17789 from zsxwing/pr17024. (cherry picked from commit 77bcd77) Signed-off-by: Shixiong Zhu <[email protected]>

[SPARK-19525][CORE] Compressing checkpoints.

7837b0c

Spark's performance improves greatly if we enable compression of checkpoints.

mridulm reviewed Feb 22, 2017

View reviewed changes

[SPARK-19525][CORE] Addressing comments.

18e7ba6

zsxwing requested changes Apr 17, 2017

View reviewed changes

zsxwing mentioned this pull request Apr 27, 2017

[SPARK-19525][CORE]Add RDD checkpoint compression support #17789

Closed

asfgit closed this in 77bcd77 Apr 28, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-19525][CORE] Compressing checkpoints. #17024

[SPARK-19525][CORE] Compressing checkpoints. #17024

aramesh117 commented Feb 22, 2017 •

edited

Loading

AmplabJenkins commented Feb 22, 2017

mridulm left a comment

mridulm Feb 22, 2017 •

edited

Loading

mridulm Feb 22, 2017

mridulm Feb 22, 2017

aramesh117 Feb 22, 2017

mridulm Feb 23, 2017

mridulm Feb 22, 2017

mridulm Feb 22, 2017

mridulm Feb 22, 2017

aramesh117 commented Feb 22, 2017

mridulm commented Feb 23, 2017

aramesh117 commented Feb 25, 2017

mridulm commented Feb 27, 2017

mridulm commented Feb 27, 2017

aramesh117 commented Mar 7, 2017

mridulm commented Mar 7, 2017

aramesh117 commented Apr 12, 2017

mridulm commented Apr 12, 2017

zsxwing left a comment

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing Apr 17, 2017

zsxwing commented Apr 26, 2017 •

edited

Loading

zsxwing commented Apr 27, 2017

		import org.apache.spark.util.{SerializableConfiguration, Utils}

[SPARK-19525][CORE] Compressing checkpoints. #17024

[SPARK-19525][CORE] Compressing checkpoints. #17024

Conversation

aramesh117 commented Feb 22, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

AmplabJenkins commented Feb 22, 2017

mridulm left a comment

Choose a reason for hiding this comment

mridulm Feb 22, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aramesh117 commented Feb 22, 2017

mridulm commented Feb 23, 2017

aramesh117 commented Feb 25, 2017

mridulm commented Feb 27, 2017

mridulm commented Feb 27, 2017

aramesh117 commented Mar 7, 2017

mridulm commented Mar 7, 2017

aramesh117 commented Apr 12, 2017

mridulm commented Apr 12, 2017

zsxwing left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing commented Apr 26, 2017 • edited Loading

zsxwing commented Apr 27, 2017

aramesh117 commented Feb 22, 2017 •

edited

Loading

mridulm Feb 22, 2017 •

edited

Loading

zsxwing commented Apr 26, 2017 •

edited

Loading