spark中spark-内部实现原理.doc

这两天在看spark-submit之后任务是怎么启动的，写篇文章记录一下自己的理解思路：SparkSubmitSparkDeploySchedulerBackendAppClienttryRegisterAllMaster1:客户端启动，初始化相关的环境变量，包括Application代码的提交2:向Master注册Driver 这里需要注意，Master，Worker是已经启动，换句话说，我们的spark集群已经启动。3: SparkDeploySchedulerBackend中启动Application。向Master注册Application def registerWithMaster() tryRegisterAllMasters() import context.dispatcher var retries = 0 registrationRetryTimer = Some context.system.scheduler.schedule(REGISTRATION_TIMEOUT, REGISTRATION_TIMEOUT) Utils.tryOrExit retries += 1 if (registered) registrationRetryTimer.foreach(_.cancel() else if (retries = REGISTRATION_RETRIES) markDead(All masters are unresponsive! Giving up.) else tryRegisterAllMasters() def tryRegisterAllMasters() for (masterAkkaUrl if (state = RecoveryState.STANDBY) / ignore, dont send response else logInfo(Registering app + ) val app = createApplication(description, sender) registerApplication(app) logInfo(Registered app + + with ID + app.id) persistenceEngine.addApplication(app) sender ! RegisteredApplication(app.id, masterUrl) schedule() 进行注册Application，并且发送已经注册App的消息。然后进行schedule()。Schedule函数如下： /* * Schedule the currently available resources among waiting apps. This method will be called * every time a new app joins or resource availability changes. */ private def schedule(): Unit = if (state != RecoveryState.ALIVE) return / Drivers take strict precedence over executors val shuffledWorkers = Random.shuffle(workers) / Randomization helps balance drivers for (worker - shuffledWorkers if worker.state = WorkerState.ALIVE) for (driver = driver.desc.mem & worker.coresFree = driver.desc.cores) launchDriver(worker, driver) waitingDrivers -= driver startExecutorsOnWorkers() 4：Master接受请求后，根据资源情况向worker发送指令启动Executor。（上面最后一行代码）private def startExecutorsOnWorkers(): Unit = / Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app / in the queue, then the second app, etc. if (spreadOutApps) / Try to spread out each app among all the workers, until it has all its cores for (app 0) val usableWorkers = workers.toArray.filter(_.state = WorkerState.ALIVE) .filter(worker = worker.memoryFree = app.desc.memoryPerExecutorMB & worker.coresFree = app.desc.coresPerExecutor.getOrElse(1) .sortBy(_.coresFree).reverse val numUsable = usableWorkers.length val assigned = new ArrayInt(numUsable) / Number of cores to give on each node var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) var pos = 0 while (toAssign 0) if (usableWorkers(pos).coresFree - assigned(pos) 0) toAssign -= 1 assigned(pos) += 1 pos = (pos + 1) % numUsable / Now that weve decided how many cores to give on each node, lets actually give them for (pos 0) allocateWorkerResourceToExecutors(app, assigned(pos), usableWorkers(pos) else / Pack each app into as few workers as possible until weve assigned all its cores for (worker 0 & worker.state = WorkerState.ALIVE) for (app 0) allocateWorkerResourceToExecutors(app, app.coresLeft, worker) 这段代码就是在worker上启动executor。代码中的注释如下：Schedule executors to be launched on the workers. * There are two modes of launching executors. The first attempts to spread out an applications* executors on as many workers as possible, while the second does the opposite (i.e. launch them on as few workers as possible). The former is usually better for data locality purposes and is the default.注释中同时指出：对于executor的启动有两种方式：第一种是在一台worker上启动最够多的executor，第二种是相反的，在worker上启动足够少的executor。这就和Application的分发刚好对应上。Schedule函数中关键代码allocateWorkerResourceToExecutors功能如下：private def allocateWorkerResourceToExecutors( app: ApplicationInfo, coresToAllocate: Int, worker: WorkerInfo): Unit = val memoryPerExecutor = app.desc.memoryPerExecutorMB val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(coresToAllocate) var coresLeft = coresToAllocate while (coresLeft = coresPerExecutor & worker.memoryFree = memoryPerExecutor) val exec = app.addExecutor(worker, coresPerExecutor) coresLeft -= coresPerExecutor launchExecutor(worker, exec) app.state = ApplicationState.RUNNING 红色字体，这下应该看明白了吧。这里启动了Executor。这里executor已经启动了，然后就等着来运行任务了。而对于任务的运行，是不是想到了DAGScheduler和TaskScheduler。5：DAGScheduler进行Job的Stage划分（这里我之前写过一篇文章，所以不再细贴代码了，大体介绍一下就可以了）这里主要是处理handleJobSubmitted事件，首先需要新建stage和activejob。Stage构建的话有几点需要注意哦。Partition和依赖。然后开始提交Stage，这里可以看到，需要先检测有没有miss的stage。执行submitMissingTasks(stage, jobId.get)。针对不同类型的stage，所做的操作是不一样的，我们的stage类型有ShuffleMapStage和ResultStage。然后就是非常重要的一点了，task的序列化。注意：DAGScheduler提交给TaskScheduler的是TaskSet，即一组相同的任务，只是处理的数据不同。OK，继续说task的序列化。ShuffleMapStage和ResultStage序列化的过程也是不一样的。序列化后的任务用广播的方式发送，每个executor得到之后首先会进行反序列化，这样在不同的executor上运行的task是隔离的，不会互相影响。然后开始构建task，ShuffleMapTask和ResultTask。构建完之后，形成TaskSet，将其提交给TaskScheduler。6：TaskScheduler向注册来的Executor发送LaunchTask命令进行任务的运行。这里解释一下，因为源码中比较绕，首先会在TaskSchedulerImpl中执行SubmitTask函数，backend.reviveOffers()。var backend: SchedulerBackend = null这里能够看到我们是调用了reviveOffers函数。然后进入到原函数中看到如下代码： override def reviveOffers() driverEndpoint.send(ReviveOffers) 这里向driverEndpoint发送了ReviveOffers消息。所以我们需要进入到driverEndpoint中找到处理ReviveOffers的函数。driverEndpoint = rpcEnv.setupEndpoint( CoarseGrainedSchedulerBackend.ENDPOINT_NAME, new DriverEndpoint(rpcEnv, properties)这里能够看到new DriverEndpoint，是在CoarseGrainedSchedulerBackend里面。OK，在这里面找ReviveOffers函数。 case ReviveOffers = makeOffers() def makeOffers() launchTasks(scheduler.resourceOffers(executorDataMap.map case (id, executorData) = new WorkerOffer(id, executorData.executorHost, executorData.freeCores) .toSeq) LaunchTasks函数开始处理了。/ Launch tasks returned by a set of resource offers def launchTasks(tasks: SeqSeqTaskDescription) for (task = akkaFrameSize - AkkaUtils.reservedSizeBytes) val taskSetId = scheduler.taskIdToTaskSetId(task.taskId) scheduler.activeTaskSets.get(taskSetId).foreach taskSet = try var msg = Serialized task %s:%d was %d bytes, which exceeds max allowed: + spark.akka.frameSize (%d bytes) - reserved (%d bytes). Consider increasing + spark.akka.frameSize or using broadcast var