private var taskScheduler: TaskScheduler = {//...
}
taskScheduler.start()
private var dagScheduler = new DAGScheduler(taskScheduler)
dagScheduler.start()

private def run() {SparkEnv.set(env)while (true) {val event = eventQueue.poll(POLL_TIMEOUT, TimeUnit.MILLISECONDS)if (event != null) {logDebug("Got event of type " + event.getClass.getName)}if (event != null) {if (processEvent(event)) {return}}val time = System.currentTimeMillis() // TODO: use a pluggable clock for testabilityif (failed.size > 0 && time > lastFetchFailureTime + RESUBMIT_TIMEOUT) {resubmitFailedStages()} else {submitWaitingStages()}}
}

val textFile = sc.textFile("README.md")
textFile.filter(line => line.contains("Spark")).count()

val textFile = sc.textFile("README.md")textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)

def runJob[T, U: ClassManifest](finalRdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],callSite: String,allowLocal: Boolean,resultHandler: (Int, U) => Unit)
{if (partitions.size == 0) {return}val (toSubmit, waiter) = prepareJob(finalRdd, func, partitions, callSite, allowLocal, resultHandler)eventQueue.put(toSubmit)waiter.awaitResult() match {case JobSucceeded => {}case JobFailed(exception: Exception) =>logInfo("Failed to run " + callSite)throw exception}
}

case JobSubmitted(finalRDD, func, partitions, allowLocal, callSite, listener) =>val runId = nextRunId.getAndIncrement()val finalStage = newStage(finalRDD, None, runId)val job = new ActiveJob(runId, finalStage, func, partitions, callSite, listener)clearCacheLocs()if (allowLocal && finalStage.parents.size == 0 && partitions.length == 1) {runLocally(job)} else {activeJobs += jobresultStageToJob(finalStage) = jobsubmitStage(finalStage)}

private def submitStage(stage: Stage) {if (!waiting(stage) && !running(stage) && !failed(stage)) {val missing = getMissingParentStages(stage).sortBy(_.id)if (missing == Nil) {submitMissingTasks(stage)running += stage} else {for (parent <- missing) {submitStage(parent)}waiting += stage}}
}

private def getMissingParentStages(stage: Stage): List[Stage] = {val missing = new HashSet[Stage]val visited = new HashSet[RDD[_]]def visit(rdd: RDD[_]) {if (!visited(rdd)) {visited += rddif (getCacheLocs(rdd).contains(Nil)) {for (dep <- rdd.dependencies) {dep match {case shufDep: ShuffleDependency[_,_] =>val mapStage = getShuffleMapStage(shufDep, stage.priority)if (!mapStage.isAvailable) {missing += mapStage}case narrowDep: NarrowDependency[_] =>visit(narrowDep.rdd)}}}}}visit(stage.rdd)missing.toList
}

private def submitMissingTasks(stage: Stage) {val myPending = pendingTasks.getOrElseUpdate(stage, new HashSet)myPending.clear()var tasks = ArrayBuffer[Task[_]]()if (stage.isShuffleMap) {for (p <- 0 until stage.numPartitions if stage.outputLocs(p) == Nil) {val locs = getPreferredLocs(stage.rdd, p)tasks += new ShuffleMapTask(stage.id, stage.rdd, stage.shuffleDep.get, p, locs)}} else {val job = resultStageToJob(stage)for (id <- 0 until job.numPartitions if (!job.finished(id))) {val partition = job.partitions(id)val locs = getPreferredLocs(stage.rdd, partition)tasks += new ResultTask(stage.id, stage.rdd, job.func, partition, locs, id)}}if (tasks.size > 0) {myPending ++= taskstaskSched.submitTasks(new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.priority))if (!stage.submissionTime.isDefined) {stage.submissionTime = Some(System.currentTimeMillis())}} else {running -= stage}
}

private def handleTaskCompletion(event: CompletionEvent) {val task = event.taskval stage = idToStage(task.stageId)def markStageAsFinished(stage: Stage) = {val serviceTime = stage.submissionTime match {case Some(t) => "%.03f".format((System.currentTimeMillis() - t) / 1000.0)case _ => "Unkown"}logInfo("%s (%s) finished in %s s".format(stage, stage.origin, serviceTime))running -= stage}event.reason match {case Success =>...task match {case rt: ResultTask[_, _] =>...case smt: ShuffleMapTask =>...}case Resubmitted =>...case FetchFailed(bmAddress, shuffleId, mapId, reduceId) =>...case other =>abortStage(idToStage(task.stageId), task + " failed: " + other)}
}

override def run(attemptId: Long): U = {val context = new TaskContext(stageId, partition, attemptId)try {func(context, rdd.iterator(split, context))} finally {context.executeOnCompleteCallbacks()}}

 override def run(attemptId: Long): MapStatus = {val numOutputSplits = dep.partitioner.numPartitionsval taskContext = new TaskContext(stageId, partition, attemptId)try {val buckets = Array.fill(numOutputSplits)(new ArrayBuffer[(Any, Any)])for (elem <- rdd.iterator(split, taskContext)) {val pair = elem.asInstanceOf[(Any, Any)]val bucketId = dep.partitioner.getPartition(pair._1)buckets(bucketId) += pair}val compressedSizes = new Array[Byte](numOutputSplits)val blockManager = SparkEnv.get.blockManagerfor (i <- 0 until numOutputSplits) {val blockId = "shuffle_" + dep.shuffleId + "_" + partition + "_" + ival iter: Iterator[(Any, Any)] = buckets(i).iteratorval size = blockManager.put(blockId, iter, StorageLevel.DISK_ONLY, false)compressedSizes(i) = MapOutputTracker.compressSize(size)}return new MapStatus(blockManager.blockManagerId, compressedSizes)} finally {taskContext.executeOnCompleteCallbacks()}}

final def iterator(split: Partition, context: TaskContext): Iterator[T] = {if (storageLevel != StorageLevel.NONE) {SparkEnv.get.cacheManager.getOrCompute(this, split, context, storageLevel)} else {computeOrReadCheckpoint(split, context)}
}
private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext): Iterator[T] = {if (isCheckpointed) {firstParent[T].iterator(split, context)} else {compute(split, context)}
} 

 master match {...case SPARK_REGEX(sparkUrl) =>val scheduler = new ClusterScheduler(this)val backend = new SparkDeploySchedulerBackend(scheduler, this, sparkUrl, appName)scheduler.initialize(backend)schedulercase LOCAL_CLUSTER_REGEX(numSlaves, coresPerSlave, memoryPerSlave) =>...case _ =>...}
}
taskScheduler.start()

override def start() {backend.start()if (System.getProperty("spark.speculation", "false") == "true") {new Thread("ClusterScheduler speculation check") {setDaemon(true)override def run() {while (true) {try {Thread.sleep(SPECULATION_INTERVAL)} catch {case e: InterruptedException => {}}checkSpeculatableTasks()}}}.start()}
}

override def start() {super.start()val driverUrl = "akka://spark@%s:%s/user/%s".format(System.getProperty("spark.driver.host"), System.getProperty("spark.driver.port"),StandaloneSchedulerBackend.ACTOR_NAME)val args = Seq(driverUrl, "", "", "")val command = Command("spark.executor.StandaloneExecutorBackend", args, sc.executorEnvs)val sparkHome = sc.getSparkHome().getOrElse(throw new IllegalArgumentException("must supply spark home for spark standalone"))val appDesc = new ApplicationDescription(appName, maxCores, executorMemory, command, sparkHome)client = new Client(sc.env.actorSystem, master, appDesc, this)client.start()
}

override def start() {val properties = new ArrayBuffer[(String, String)]val iterator = System.getProperties.entrySet.iteratorwhile (iterator.hasNext) {val entry = iterator.nextval (key, value) = (entry.getKey.toString, entry.getValue.toString)if (key.startsWith("spark.")) {properties += ((key, value))}}driverActor = actorSystem.actorOf(Props(new DriverActor(properties)), name = StandaloneSchedulerBackend.ACTOR_NAME)
}

override def submitTasks(taskSet: TaskSet) {val tasks = taskSet.taskslogInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")this.synchronized {val manager = new TaskSetManager(this, taskSet)activeTaskSets(taskSet.id) = manageractiveTaskSetsQueue += managertaskSetTaskIds(taskSet.id) = new HashSet[Long]()if (hasReceivedTask == false) {starvationTimer.scheduleAtFixedRate(new TimerTask() {override def run() {if (!hasLaunchedTask) {logWarning("Initial job has not accepted any resources; " +"check your cluster UI to ensure that workers are registered")} else {this.cancel()}}}, STARVATION_TIMEOUT, STARVATION_TIMEOUT)}hasReceivedTask = true;}backend.reviveOffers()
}

// Make fake resource offers on just one executor
def makeOffers(executorId: String) {launchTasks(scheduler.resourceOffers(Seq(new WorkerOffer(executorId, executorHost(executorId), freeCores(executorId)))))
}
// Launch tasks returned by a set of resource offers
def launchTasks(tasks: Seq[Seq[TaskDescription]]) {for (task <- tasks.flatten) {freeCores(task.executorId) -= 1executorActor(task.executorId) ! LaunchTask(task)}
}

override def receive = {case RegisteredExecutor(sparkProperties) =>...  case RegisterExecutorFailed(message) =>...case LaunchTask(taskDesc) =>logInfo("Got assigned task " + taskDesc.taskId)executor.launchTask(this, taskDesc.taskId, taskDesc.serializedTask)case Terminated(_) | RemoteClientDisconnected(_, _) | RemoteClientShutdown(_, _) =>...
}
def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) {threadPool.execute(new TaskRunner(context, taskId, serializedTask))
}

class TaskRunner(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer)extends Runnable {override def run() {SparkEnv.set(env)Thread.currentThread.setContextClassLoader(urlClassLoader)val ser = SparkEnv.get.closureSerializer.newInstance()logInfo("Running task ID " + taskId)context.statusUpdate(taskId, TaskState.RUNNING, EMPTY_BYTE_BUFFER)try {SparkEnv.set(env)Accumulators.clear()val (taskFiles, taskJars, taskBytes) = Task.deserializeWithDependencies(serializedTask)updateDependencies(taskFiles, taskJars)val task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.getContextClassLoader)logInfo("Its generation is " + task.generation)env.mapOutputTracker.updateGeneration(task.generation)val value = task.run(taskId.toInt)val accumUpdates = Accumulators.valuesval result = new TaskResult(value, accumUpdates)val serializedResult = ser.serialize(result)logInfo("Serialized size of result for " + taskId + " is " + serializedResult.limit)context.statusUpdate(taskId, TaskState.FINISHED, serializedResult)logInfo("Finished task ID " + taskId)} catch {case ffe: FetchFailedException => {val reason = ffe.toTaskEndReasoncontext.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))}case t: Throwable => {val reason = ExceptionFailure(t)context.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))// TODO: Should we exit the whole executor here? On the one hand, the failed task may// have left some weird state around depending on when the exception was thrown, but on// the other hand, maybe we could detect that when future tasks fail and exit then.logError("Exception in task ID " + taskId, t)//System.exit(1)}}}
}