diff options
| author | Vadim Chekan <kot.begemot@gmail.com> | 2013-09-20 12:13:48 -0700 |
|---|---|---|
| committer | Vadim Chekan <kot.begemot@gmail.com> | 2013-09-20 12:13:48 -0700 |
| commit | fbe40c5806a01e38c56b12a09bc4b84681a99602 (patch) | |
| tree | 0cdfa8f5b420effdbf1bce233b4b9b26b6ee5ad8 /streaming | |
| parent | a106ed8b97e707b36818c11d1d7211fa28636178 (diff) | |
| download | spark-fbe40c5806a01e38c56b12a09bc4b84681a99602.tar.gz spark-fbe40c5806a01e38c56b12a09bc4b84681a99602.tar.bz2 spark-fbe40c5806a01e38c56b12a09bc4b84681a99602.zip | |
Serialize and restore spark.cleaner.ttl to savepoint
Diffstat (limited to 'streaming')
| -rw-r--r-- | streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala | 2 | ||||
| -rw-r--r-- | streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala | 4 |
2 files changed, 6 insertions, 0 deletions
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala b/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala index 2d8f072624..bb9febad38 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala @@ -26,6 +26,7 @@ import org.apache.hadoop.conf.Configuration import org.apache.spark.Logging import org.apache.spark.io.CompressionCodec +import org.apache.spark.util.MetadataCleaner private[streaming] @@ -40,6 +41,7 @@ class Checkpoint(@transient ssc: StreamingContext, val checkpointTime: Time) val checkpointDir = ssc.checkpointDir val checkpointDuration = ssc.checkpointDuration val pendingTimes = ssc.scheduler.jobManager.getPendingTimes() + val delaySeconds = MetadataCleaner.getDelaySeconds def validate() { assert(master != null, "Checkpoint.master is null") diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala index 878725c705..098081d245 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala @@ -100,6 +100,10 @@ class StreamingContext private ( "both SparkContext and checkpoint as null") } + if(cp_ != null && cp_.delaySeconds >= 0 && MetadataCleaner.getDelaySeconds < 0) { + MetadataCleaner.setDelaySeconds(cp_.delaySeconds) + } + if (MetadataCleaner.getDelaySeconds < 0) { throw new SparkException("Spark Streaming cannot be used without setting spark.cleaner.ttl; " + "set this property before creating a SparkContext (use SPARK_JAVA_OPTS for the shell)") |