You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pw...@apache.org on 2013/12/31 19:13:05 UTC
[02/16] git commit: Merge branch 'scheduler-update' into
filestream-fix
Merge branch 'scheduler-update' into filestream-fix
Conflicts:
core/src/main/scala/org/apache/spark/rdd/CheckpointRDD.scala
streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala
streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala
streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/984c5824
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/984c5824
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/984c5824
Branch: refs/heads/master
Commit: 984c5824876e0daceb8a74af57593926faf727ce
Parents: 5e9ce83 ec71b44
Author: Tathagata Das <ta...@gmail.com>
Authored: Thu Dec 19 11:20:48 2013 -0800
Committer: Tathagata Das <ta...@gmail.com>
Committed: Thu Dec 19 11:20:48 2013 -0800
----------------------------------------------------------------------
.gitignore | 1 +
README.md | 14 +-
assembly/pom.xml | 16 +-
bagel/pom.xml | 12 +-
bin/compute-classpath.cmd | 2 +-
bin/compute-classpath.sh | 2 +-
core/pom.xml | 35 +-
.../apache/spark/network/netty/FileClient.java | 2 -
.../apache/spark/network/netty/FileServer.java | 1 -
.../org/apache/spark/MapOutputTracker.scala | 29 +-
.../scala/org/apache/spark/Partitioner.scala | 8 +-
.../scala/org/apache/spark/SparkContext.scala | 77 +-
.../main/scala/org/apache/spark/SparkEnv.scala | 13 +-
.../main/scala/org/apache/spark/TaskState.scala | 3 +-
.../apache/spark/api/java/JavaDoubleRDD.scala | 9 +-
.../org/apache/spark/api/java/JavaPairRDD.scala | 77 +-
.../org/apache/spark/api/java/JavaRDD.scala | 7 +-
.../org/apache/spark/api/java/JavaRDDLike.scala | 32 +-
.../spark/api/java/JavaSparkContext.scala | 60 +-
.../java/JavaSparkContextVarargsWorkaround.java | 1 -
.../api/java/function/FlatMapFunction.scala | 4 +-
.../api/java/function/FlatMapFunction2.scala | 4 +-
.../spark/api/java/function/Function.java | 8 +-
.../spark/api/java/function/Function2.java | 8 +-
.../spark/api/java/function/Function3.java | 8 +-
.../api/java/function/PairFlatMapFunction.java | 12 +-
.../spark/api/java/function/PairFunction.java | 12 +-
.../org/apache/spark/api/python/PythonRDD.scala | 10 +-
.../spark/api/python/PythonWorkerFactory.scala | 4 +-
.../org/apache/spark/deploy/ExecutorState.scala | 3 +-
.../apache/spark/deploy/LocalSparkCluster.scala | 13 +-
.../org/apache/spark/deploy/client/Client.scala | 48 +-
.../spark/deploy/master/ApplicationState.scala | 3 +-
.../master/FileSystemPersistenceEngine.scala | 6 +-
.../org/apache/spark/deploy/master/Master.scala | 69 +-
.../spark/deploy/master/RecoveryState.scala | 4 +-
.../spark/deploy/master/WorkerState.scala | 4 +-
.../master/ZooKeeperPersistenceEngine.scala | 6 +-
.../deploy/master/ui/ApplicationPage.scala | 5 +-
.../spark/deploy/master/ui/IndexPage.scala | 4 +-
.../spark/deploy/master/ui/MasterWebUI.scala | 2 +-
.../org/apache/spark/deploy/worker/Worker.scala | 48 +-
.../spark/deploy/worker/ui/IndexPage.scala | 5 +-
.../spark/deploy/worker/ui/WorkerWebUI.scala | 13 +-
.../executor/CoarseGrainedExecutorBackend.scala | 23 +-
.../org/apache/spark/executor/Executor.scala | 2 +-
.../spark/network/ConnectionManager.scala | 8 +-
.../spark/network/ConnectionManagerTest.scala | 4 +-
.../org/apache/spark/rdd/AsyncRDDActions.scala | 3 +-
.../scala/org/apache/spark/rdd/BlockRDD.scala | 4 +-
.../org/apache/spark/rdd/CartesianRDD.scala | 3 +-
.../org/apache/spark/rdd/CheckpointRDD.scala | 13 +-
.../org/apache/spark/rdd/CoalescedRDD.scala | 3 +-
.../apache/spark/rdd/DoubleRDDFunctions.scala | 5 +-
.../scala/org/apache/spark/rdd/EmptyRDD.scala | 5 +-
.../org/apache/spark/rdd/FilteredRDD.scala | 3 +-
.../org/apache/spark/rdd/FlatMappedRDD.scala | 3 +-
.../scala/org/apache/spark/rdd/GlommedRDD.scala | 3 +-
.../scala/org/apache/spark/rdd/JdbcRDD.scala | 4 +-
.../org/apache/spark/rdd/MapPartitionsRDD.scala | 4 +-
.../scala/org/apache/spark/rdd/MappedRDD.scala | 4 +-
.../apache/spark/rdd/OrderedRDDFunctions.scala | 10 +-
.../org/apache/spark/rdd/PairRDDFunctions.scala | 33 +-
.../spark/rdd/ParallelCollectionRDD.scala | 8 +-
.../apache/spark/rdd/PartitionPruningRDD.scala | 6 +-
.../scala/org/apache/spark/rdd/PipedRDD.scala | 3 +-
.../main/scala/org/apache/spark/rdd/RDD.scala | 64 +-
.../apache/spark/rdd/RDDCheckpointData.scala | 4 +-
.../scala/org/apache/spark/rdd/SampledRDD.scala | 5 +-
.../spark/rdd/SequenceFileRDDFunctions.scala | 11 +-
.../org/apache/spark/rdd/ShuffledRDD.scala | 6 +-
.../org/apache/spark/rdd/SubtractedRDD.scala | 5 +-
.../scala/org/apache/spark/rdd/UnionRDD.scala | 7 +-
.../apache/spark/rdd/ZippedPartitionsRDD.scala | 9 +-
.../scala/org/apache/spark/rdd/ZippedRDD.scala | 6 +-
.../apache/spark/scheduler/DAGScheduler.scala | 13 +-
.../apache/spark/scheduler/SchedulingMode.scala | 2 +-
.../apache/spark/scheduler/SparkListener.scala | 7 +-
.../spark/scheduler/SparkListenerBus.scala | 1 -
.../apache/spark/scheduler/TaskLocality.scala | 4 +-
.../scheduler/cluster/ClusterScheduler.scala | 5 +-
.../cluster/ClusterTaskSetManager.scala | 6 +-
.../cluster/CoarseGrainedSchedulerBackend.scala | 23 +-
.../cluster/SimrSchedulerBackend.scala | 2 +-
.../cluster/SparkDeploySchedulerBackend.scala | 2 +-
.../scheduler/cluster/TaskResultGetter.scala | 4 +-
.../mesos/CoarseMesosSchedulerBackend.scala | 2 +-
.../org/apache/spark/storage/BlockManager.scala | 7 +-
.../spark/storage/BlockManagerMaster.scala | 16 +-
.../spark/storage/BlockManagerMasterActor.scala | 7 +-
.../apache/spark/storage/ThreadingTest.scala | 2 +-
.../apache/spark/ui/jobs/JobProgressUI.scala | 2 +-
.../spark/ui/storage/BlockManagerUI.scala | 2 +-
.../scala/org/apache/spark/util/AkkaUtils.scala | 79 +-
.../spark/util/BoundedPriorityQueue.scala | 2 +
.../spark/util/IndestructibleActorSystem.scala | 68 +
.../org/apache/spark/util/MetadataCleaner.scala | 3 +-
.../apache/spark/util/TimeStampedHashMap.scala | 2 +-
.../scala/org/apache/spark/util/Utils.scala | 5 +-
.../spark/util/collection/OpenHashMap.scala | 3 +-
.../spark/util/collection/OpenHashSet.scala | 11 +-
.../collection/PrimitiveKeyOpenHashMap.scala | 7 +-
.../spark/util/collection/PrimitiveVector.scala | 4 +-
.../org/apache/spark/AccumulatorSuite.scala | 32 +-
.../org/apache/spark/CheckpointSuite.scala | 5 +-
.../org/apache/spark/DistributedSuite.scala | 5 +-
.../scala/org/apache/spark/DriverSuite.scala | 2 +-
.../apache/spark/MapOutputTrackerSuite.scala | 14 +-
.../scala/org/apache/spark/UnpersistSuite.scala | 2 +-
.../scala/org/apache/spark/rdd/RDDSuite.scala | 8 +-
.../spark/scheduler/SparkListenerSuite.scala | 2 +-
.../cluster/ClusterTaskSetManagerSuite.scala | 2 +-
.../cluster/TaskResultGetterSuite.scala | 4 +-
.../org/apache/spark/storage/BlockIdSuite.scala | 2 +-
.../spark/storage/BlockManagerSuite.scala | 2 +-
.../scala/org/apache/spark/ui/UISuite.scala | 1 -
.../apache/spark/util/SizeEstimatorSuite.scala | 72 +-
docs/_config.yml | 2 +-
docs/_plugins/copy_api_dirs.rb | 2 +-
docs/configuration.md | 23 +-
docs/running-on-yarn.md | 12 +-
docs/scala-programming-guide.md | 2 +-
ec2/spark_ec2.py | 4 +-
examples/pom.xml | 26 +-
.../org/apache/spark/examples/JavaLogQuery.java | 2 +-
.../org/apache/spark/examples/JavaPageRank.java | 3 +-
.../apache/spark/examples/JavaWordCount.java | 2 +-
.../apache/spark/mllib/examples/JavaALS.java | 1 -
.../streaming/examples/ActorWordCount.scala | 7 +-
.../streaming/examples/ZeroMQWordCount.scala | 8 +-
mllib/pom.xml | 12 +-
.../spark/mllib/util/MFDataGenerator.scala | 2 +-
.../spark/mllib/clustering/JavaKMeansSuite.java | 4 +-
.../mllib/recommendation/JavaALSSuite.java | 2 -
new-yarn/pom.xml | 6 +-
.../spark/deploy/yarn/ClientArguments.scala | 1 +
.../spark/deploy/yarn/WorkerLauncher.scala | 13 +-
.../deploy/yarn/YarnAllocationHandler.scala | 2 +-
pom.xml | 179 +-
project/SparkBuild.scala | 135 +-
pyspark | 10 +-
pyspark2.cmd | 2 +-
python/pyspark/rdd.py | 4 +-
repl-bin/pom.xml | 8 +-
repl-bin/src/deb/bin/run | 2 +-
repl/lib/scala-jline.jar | Bin 158463 -> 0 bytes
repl/pom.xml | 18 +-
.../main/scala/org/apache/spark/repl/Main.scala | 8 +-
.../org/apache/spark/repl/SparkExprTyper.scala | 109 ++
.../org/apache/spark/repl/SparkILoop.scala | 944 +++++-----
.../org/apache/spark/repl/SparkILoopInit.scala | 143 ++
.../org/apache/spark/repl/SparkIMain.scala | 1681 ++++++++++--------
.../org/apache/spark/repl/SparkISettings.scala | 63 -
.../org/apache/spark/repl/SparkImports.scala | 108 +-
.../spark/repl/SparkJLineCompletion.scala | 206 ++-
.../apache/spark/repl/SparkJLineReader.scala | 65 +-
.../apache/spark/repl/SparkMemberHandlers.scala | 109 +-
.../scala/org/apache/spark/repl/ReplSuite.scala | 178 +-
run-example | 2 +-
run-example2.cmd | 2 +-
spark-class | 6 +-
streaming/pom.xml | 22 +-
.../org/apache/spark/streaming/Checkpoint.scala | 2 +-
.../org/apache/spark/streaming/DStream.scala | 48 +-
.../spark/streaming/DStreamCheckpointData.scala | 6 +-
.../apache/spark/streaming/DStreamGraph.scala | 1 +
.../scala/org/apache/spark/streaming/Job.scala | 41 -
.../org/apache/spark/streaming/JobManager.scala | 88 -
.../spark/streaming/NetworkInputTracker.scala | 174 --
.../spark/streaming/PairDStreamFunctions.scala | 63 +-
.../org/apache/spark/streaming/Scheduler.scala | 131 --
.../spark/streaming/StreamingContext.scala | 80 +-
.../spark/streaming/api/java/JavaDStream.scala | 8 +-
.../streaming/api/java/JavaDStreamLike.scala | 79 +-
.../streaming/api/java/JavaPairDStream.scala | 93 +-
.../api/java/JavaStreamingContext.scala | 116 +-
.../dstream/ConstantInputDStream.scala | 3 +-
.../streaming/dstream/FileInputDStream.scala | 10 +-
.../streaming/dstream/FilteredDStream.scala | 3 +-
.../dstream/FlatMapValuedDStream.scala | 3 +-
.../streaming/dstream/FlatMappedDStream.scala | 3 +-
.../streaming/dstream/FlumeInputDStream.scala | 3 +-
.../streaming/dstream/ForEachDStream.scala | 6 +-
.../streaming/dstream/GlommedDStream.scala | 3 +-
.../spark/streaming/dstream/InputDStream.scala | 4 +-
.../streaming/dstream/KafkaInputDStream.scala | 23 +-
.../streaming/dstream/MQTTInputDStream.scala | 3 +-
.../dstream/MapPartitionedDStream.scala | 3 +-
.../streaming/dstream/MapValuedDStream.scala | 3 +-
.../spark/streaming/dstream/MappedDStream.scala | 3 +-
.../streaming/dstream/NetworkInputDStream.scala | 14 +-
.../dstream/PluggableInputDStream.scala | 3 +-
.../streaming/dstream/QueueInputDStream.scala | 4 +-
.../streaming/dstream/RawInputDStream.scala | 4 +-
.../dstream/ReducedWindowedDStream.scala | 9 +-
.../streaming/dstream/ShuffledDStream.scala | 3 +-
.../streaming/dstream/SocketInputDStream.scala | 6 +-
.../spark/streaming/dstream/StateDStream.scala | 4 +-
.../streaming/dstream/TransformedDStream.scala | 3 +-
.../spark/streaming/dstream/UnionDStream.scala | 5 +-
.../streaming/dstream/WindowedDStream.scala | 7 +-
.../streaming/receivers/ActorReceiver.scala | 35 +-
.../streaming/receivers/ZeroMQReceiver.scala | 13 +-
.../spark/streaming/scheduler/BatchInfo.scala | 37 +
.../apache/spark/streaming/scheduler/Job.scala | 41 +
.../streaming/scheduler/JobGenerator.scala | 132 ++
.../streaming/scheduler/JobScheduler.scala | 106 ++
.../spark/streaming/scheduler/JobSet.scala | 61 +
.../scheduler/NetworkInputTracker.scala | 175 ++
.../streaming/scheduler/StreamingListener.scala | 75 +
.../scheduler/StreamingListenerBus.scala | 81 +
.../streaming/util/MasterFailureTest.scala | 45 +-
.../apache/spark/streaming/JavaAPISuite.java | 88 +-
.../apache/spark/streaming/JavaTestUtils.scala | 22 +-
.../spark/streaming/BasicOperationsSuite.scala | 12 -
.../spark/streaming/CheckpointSuite.scala | 62 +-
.../apache/spark/streaming/FailureSuite.scala | 13 +-
.../spark/streaming/InputStreamsSuite.scala | 12 -
.../streaming/StreamingListenerSuite.scala | 71 +
.../apache/spark/streaming/TestSuiteBase.scala | 61 +-
.../spark/streaming/WindowOperationsSuite.scala | 14 +-
tools/pom.xml | 12 +-
.../tools/JavaAPICompletenessChecker.scala | 4 +-
yarn/pom.xml | 10 +-
.../spark/deploy/yarn/ClientArguments.scala | 1 +
.../spark/deploy/yarn/WorkerLauncher.scala | 17 +-
.../deploy/yarn/YarnAllocationHandler.scala | 4 +-
227 files changed, 4357 insertions(+), 3471 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/core/src/main/scala/org/apache/spark/api/java/JavaSparkContext.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/core/src/main/scala/org/apache/spark/rdd/CheckpointRDD.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/spark/rdd/CheckpointRDD.scala
index ef4057e,a712ef1..80385fc
--- a/core/src/main/scala/org/apache/spark/rdd/CheckpointRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/CheckpointRDD.scala
@@@ -17,16 -17,13 +17,15 @@@
package org.apache.spark.rdd
+ import java.io.IOException
+
+ import scala.reflect.ClassTag
-
-import org.apache.hadoop.fs.Path
++import java.io.{IOException}
import org.apache.spark._
++import org.apache.spark.broadcast.Broadcast
import org.apache.spark.deploy.SparkHadoopUtil
- import org.apache.hadoop.mapred.{FileInputFormat, SequenceFileInputFormat, JobConf, Reporter}
+import org.apache.hadoop.conf.Configuration
- import org.apache.hadoop.io.{NullWritable, BytesWritable}
- import org.apache.hadoop.util.ReflectionUtils
+import org.apache.hadoop.fs.Path
- import java.io.{File, IOException, EOFException}
- import java.text.NumberFormat
- import org.apache.spark.broadcast.Broadcast
private[spark] class CheckpointRDDPartition(val index: Int) extends Partition {}
@@@ -34,11 -31,9 +33,11 @@@
* This RDD represents a RDD checkpoint file (similar to HadoopRDD).
*/
private[spark]
- class CheckpointRDD[T: ClassManifest](sc: SparkContext, val checkpointPath: String)
+ class CheckpointRDD[T: ClassTag](sc: SparkContext, val checkpointPath: String)
extends RDD[T](sc, Nil) {
+ val broadcastedConf = sc.broadcast(new SerializableWritable(sc.hadoopConfiguration))
+
@transient val fs = new Path(checkpointPath).getFileSystem(sc.hadoopConfiguration)
override def getPartitions: Array[Partition] = {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/core/src/main/scala/org/apache/spark/rdd/RDDCheckpointData.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
----------------------------------------------------------------------
diff --cc streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
index d6fc2a1,41da028..71065f9
--- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
@@@ -20,17 -20,17 +20,19 @@@ package org.apache.spark.streamin
import akka.actor.Props
import akka.actor.SupervisorStrategy
import akka.zeromq.Subscribe
++import akka.util.ByteString
import org.apache.spark.streaming.dstream._
++import org.apache.spark.streaming.scheduler.StreamingListener
import org.apache.spark._
import org.apache.spark.rdd.RDD
--import org.apache.spark.streaming.receivers.ActorReceiver
import org.apache.spark.streaming.receivers.ReceiverSupervisorStrategy
import org.apache.spark.streaming.receivers.ZeroMQReceiver
import org.apache.spark.storage.StorageLevel
import org.apache.spark.util.MetadataCleaner
import org.apache.spark.streaming.receivers.ActorReceiver
++import org.apache.spark.streaming.scheduler.{JobScheduler, NetworkInputTracker}
import scala.collection.mutable.Queue
import scala.collection.Map
@@@ -37,17 -38,17 +40,17 @@@ import scala.reflect.ClassTa
import java.io.InputStream
import java.util.concurrent.atomic.AtomicInteger
--import java.util.UUID
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text
import org.apache.hadoop.mapreduce.{InputFormat => NewInputFormat}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.fs.Path
++
import twitter4j.Status
import twitter4j.auth.Authorization
- import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.streaming.scheduler._
-import akka.util.ByteString
++
+
/**
* A StreamingContext is the main entry point for Spark Streaming functionality. Besides the basic
@@@ -599,9 -599,8 +604,4 @@@ object StreamingContext
prefix + "-" + time.milliseconds + "." + suffix
}
}
- /*
- protected[streaming] def getSparkCheckpointDir(sscCheckpointDir: String): String = {
- new Path(sscCheckpointDir, UUID.randomUUID.toString).toString
- }
- */
-
- protected[streaming] def getSparkCheckpointDir(sscCheckpointDir: String): String = {
- new Path(sscCheckpointDir, UUID.randomUUID.toString).toString
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala
----------------------------------------------------------------------
diff --cc streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala
index 1a8db3a,39e2523..7951354
--- a/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala
@@@ -48,13 -50,13 +50,13 @@@ class FileInputDStream[K: ClassTag, V:
override def start() {
if (newFilesOnly) {
- lastModTime = graph.zeroTime.milliseconds
+ prevModTime = graph.zeroTime.milliseconds
} else {
- lastModTime = 0
+ prevModTime = 0
}
- logDebug("LastModTime initialized to " + lastModTime + ", new files only = " + newFilesOnly)
+ logDebug("LastModTime initialized to " + prevModTime + ", new files only = " + newFilesOnly)
}
-
+
override def stop() { }
/**
@@@ -222,8 -195,5 +224,6 @@@
private[streaming]
object FileInputDStream {
+ val MAX_ATTEMPTS = 10
def defaultFilter(path: Path): Boolean = !path.getName().startsWith(".")
}
-
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala
----------------------------------------------------------------------
diff --cc streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala
index 0000000,1cd0b9b..cf3fc82
mode 000000,100644..100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala
@@@ -1,0 -1,131 +1,132 @@@
+ /*
+ * 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.streaming.scheduler
+
+ import org.apache.spark.SparkEnv
+ import org.apache.spark.Logging
+ import org.apache.spark.streaming.{Checkpoint, Time, CheckpointWriter}
+ import org.apache.spark.streaming.util.{ManualClock, RecurringTimer, Clock}
+
+ /**
+ * This class generates jobs from DStreams as well as drives checkpointing and cleaning
+ * up DStream metadata.
+ */
+ private[streaming]
+ class JobGenerator(jobScheduler: JobScheduler) extends Logging {
+
+ initLogging()
++
+ val ssc = jobScheduler.ssc
+ val clockClass = System.getProperty(
+ "spark.streaming.clock", "org.apache.spark.streaming.util.SystemClock")
+ val clock = Class.forName(clockClass).newInstance().asInstanceOf[Clock]
+ val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds,
+ longTime => generateJobs(new Time(longTime)))
+ val graph = ssc.graph
+ lazy val checkpointWriter = if (ssc.checkpointDuration != null && ssc.checkpointDir != null) {
- new CheckpointWriter(ssc.checkpointDir)
++ new CheckpointWriter(ssc.checkpointDir, ssc.sparkContext.hadoopConfiguration)
+ } else {
+ null
+ }
+
+ var latestTime: Time = null
+
+ def start() = synchronized {
+ if (ssc.isCheckpointPresent) {
+ restart()
+ } else {
+ startFirstTime()
+ }
+ logInfo("JobGenerator started")
+ }
+
+ def stop() = synchronized {
+ timer.stop()
+ if (checkpointWriter != null) checkpointWriter.stop()
+ ssc.graph.stop()
+ logInfo("JobGenerator stopped")
+ }
+
+ private def startFirstTime() {
+ val startTime = new Time(timer.getStartTime())
+ graph.start(startTime - graph.batchDuration)
+ timer.start(startTime.milliseconds)
+ logInfo("JobGenerator's timer started at " + startTime)
+ }
+
+ private def restart() {
+ // If manual clock is being used for testing, then
+ // either set the manual clock to the last checkpointed time,
+ // or if the property is defined set it to that time
+ if (clock.isInstanceOf[ManualClock]) {
+ val lastTime = ssc.initialCheckpoint.checkpointTime.milliseconds
+ val jumpTime = System.getProperty("spark.streaming.manualClock.jump", "0").toLong
+ clock.asInstanceOf[ManualClock].setTime(lastTime + jumpTime)
+ }
+
+ val batchDuration = ssc.graph.batchDuration
+
+ // Batches when the master was down, that is,
+ // between the checkpoint and current restart time
+ val checkpointTime = ssc.initialCheckpoint.checkpointTime
+ val restartTime = new Time(timer.getRestartTime(graph.zeroTime.milliseconds))
+ val downTimes = checkpointTime.until(restartTime, batchDuration)
+ logInfo("Batches during down time: " + downTimes.mkString(", "))
+
+ // Batches that were unprocessed before failure
+ val pendingTimes = ssc.initialCheckpoint.pendingTimes
+ logInfo("Batches pending processing: " + pendingTimes.mkString(", "))
+ // Reschedule jobs for these times
+ val timesToReschedule = (pendingTimes ++ downTimes).distinct.sorted(Time.ordering)
+ logInfo("Batches to reschedule: " + timesToReschedule.mkString(", "))
+ timesToReschedule.foreach(time =>
+ jobScheduler.runJobs(time, graph.generateJobs(time))
+ )
+
+ // Restart the timer
+ timer.start(restartTime.milliseconds)
+ logInfo("JobGenerator's timer restarted at " + restartTime)
+ }
+
+ /** Generate jobs and perform checkpoint for the given `time`. */
+ private def generateJobs(time: Time) {
+ SparkEnv.set(ssc.env)
+ logInfo("\n-----------------------------------------------------\n")
+ jobScheduler.runJobs(time, graph.generateJobs(time))
+ latestTime = time
+ doCheckpoint(time)
+ }
+
+ /**
+ * On batch completion, clear old metadata and checkpoint computation.
+ */
+ private[streaming] def onBatchCompletion(time: Time) {
+ ssc.graph.clearOldMetadata(time)
+ doCheckpoint(time)
+ }
+
+ /** Perform checkpoint for the give `time`. */
+ private def doCheckpoint(time: Time) = synchronized {
+ if (checkpointWriter != null && (time - graph.zeroTime).isMultipleOf(ssc.checkpointDuration)) {
+ logInfo("Checkpointing graph for time " + time)
+ ssc.graph.updateCheckpointData(time)
+ checkpointWriter.write(new Checkpoint(ssc, time))
+ }
+ }
+ }
+
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/984c5824/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
----------------------------------------------------------------------
diff --cc streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
index e517549,67a0841..4e25c95
--- a/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
@@@ -17,20 -17,24 +17,18 @@@
package org.apache.spark.streaming
--import dstream.FileInputDStream
--import org.apache.spark.streaming.StreamingContext._
import java.io.File
- import runtime.RichInt
- import org.scalatest.BeforeAndAfter
+
+ import scala.collection.mutable.ArrayBuffer
+ import scala.reflect.ClassTag
-
import org.apache.commons.io.FileUtils
- import collection.mutable.{SynchronizedBuffer, ArrayBuffer}
- import util.{Clock, ManualClock}
-import org.scalatest.BeforeAndAfter
-
import com.google.common.io.Files
-
-import org.apache.spark.streaming.StreamingContext.toPairDStreamFunctions
+import org.apache.hadoop.fs.{Path, FileSystem}
+import org.apache.hadoop.conf.Configuration
-
-
++import org.apache.spark.streaming.StreamingContext._
+ import org.apache.spark.streaming.dstream.FileInputDStream
+ import org.apache.spark.streaming.util.ManualClock
-
-
/**
* This test suites tests the checkpointing functionality of DStreams -
* the checkpointing of a DStream's RDDs as well as the checkpointing of