数据处理-Spark Worker工作流程图启动Driver源码解读

下载Jar文件，返回Jar在本地的路径，将程序打成JAR包上传到HDFS上，这样每台机器均可以从HDFS上下载。

/**

* Download the user jar into the supplied directory and return its local path.

* Will throw an exception if there are errors downloading the jar.

*/

private def downloadUserJar(driverDir: File): String = {

//

??val jarPath = new Path(driverDesc.jarUrl)

//从HDFS上获取Jar文件。

??val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)

??val destPath = new File(driverDir.getAbsolutePath,jarPath.getName)

??val jarFileName = jarPath.getName

??val localJarFile = new File(driverDir,jarFileName)

??val localJarFilename = localJarFile.getAbsolutePath

??if (!localJarFile.exists()) { // May already exist if running multiple workers on one node

? ? logInfo(s"Copying user jar?jarPathtodestPath")

? ? Utils.fetchFile(

? ?? ?driverDesc.jarUrl,

? ?? ?driverDir,

? ?? ?conf,

? ?? ?securityManager,

? ?? ?hadoopConf,

? ?? ?System.currentTimeMillis(),

? ?? ?useCache = false)

??}

??if (!localJarFile.exists()) { // Verify copy succeeded

? ? throw new Exception(s"Did not see expected jar?jarFileNameindriverDir")

??}

??localJarFilename

}

有些变量在开始的时候是占位符，因为还没有初始化，所以在实际运行的时候要初始化。

def substituteVariables(argument: String): String = argument match {

??case "{{WORKER_URL}}" => workerUrl? ?? ???

??case "{{USER_JAR}}" => localJarFilename //前面已经下载好了。

??case other => other

}

command主要就是构建进程执行类的入口

// TODO: If we add ability to submit multiple jars they should also be added here

// driverDesc.command指定启动的时候运行什么类。

??val builder = CommandUtils.buildProcessBuilder(driverDesc.command,securityManager,

? ? driverDesc.mem,sparkHome.getAbsolutePath,substituteVariables)

//launchDriver

??launchDriver(builder,driverDir,driverDesc.supervise)

}

launchDriver的源码如下：将stdout和stderr重定向到了baseDir之下了，这样就可以通过log去查看之前的执行情况

private def launchDriver(builder: ProcessBuilder,baseDir: File,supervise: Boolean) {

??builder.directory(baseDir)

??def initialize(process: Process): Unit = {

? ? // Redirect stdout and stderr to files

? ? val stdout = new File(baseDir,"stdout")

? ? CommandUtils.redirectStream(process.getInputStream,stdout)

? ? val stderr = new File(baseDir,"stderr")

? ?? ???//将command格式化一下

? ? val formattedCommand = builder.command.asScala.mkString(""","" "",""")

? ? val header = "Launch Command: %sn%snn".format(formattedCommand,"=" * 40)

? ? Files.append(header,stderr,UTF_8)

? ? CommandUtils.redirectStream(process.getErrorStream,stderr)

??}

??runCommandWithRetry(ProcessBuilderLike(builder),initialize,supervise)

}

ProcessBuilderLike静态方法:

private[deploy] object ProcessBuilderLike {

//apply方法复写了start方法

??def apply(processBuilder: ProcessBuilder): ProcessBuilderLike = new ProcessBuilderLike {

? ? override def start(): Process = processBuilder.start()

? ? override def command: Seq[String] = processBuilder.command().asScala

??}

}

ProcessBuilderLike源码如下：

// Needed because ProcessBuilder is a final class and cannot be mocked

private[deploy] trait ProcessBuilderLike {

??def start(): Process

??def command: Seq[String]

}

而在runCommandWithRetry方法中：

//传入ProcessBuilderLike的接口

def runCommandWithRetry(

? ? command: ProcessBuilderLike,initialize: Process => Unit,supervise: Boolean): Unit = {

??// Time to wait between submission retries.

??var waitSeconds = 1

??// A run of this many seconds resets the exponential back-off.

??val successfulRunDuration = 5

??var keepTrying = !killed

??while (keepTrying) {

? ? logInfo("Launch Command: " + command.command.mkString(""","""))

? ? synchronized {

? ?? ?if (killed) { return }

? ?? ???//调用ProcessBuilderLike的start()方法

? ?? ?process = Some(command.start())

? ?? ?initialize(process.get)

? ? }

? ? val processStart = clock.getTimeMillis()

//然后再调用process.get.waitFor()来完成启动Driver。

? ? val exitCode = process.get.waitFor()

? ? if (clock.getTimeMillis() - processStart > successfulRunDuration * 1000) {

? ?? ?waitSeconds = 1

? ? }

??if (supervise && exitCode != 0 && !killed) {

? ? logInfo(s"Command exited with status?exitCode,re?launchingafterwaitSeconds s.")

? ? sleeper.sleep(waitSeconds)

? ? waitSeconds = waitSeconds * 2 // exponential back-off

??}

??keepTrying = supervise && exitCode != 0 && !killed

??finalExitCode = Some(exitCode)

}

如果Driver的状态有变，则会给自己发条消息

worker.send(DriverStateChanged(driverId,state,finalException))

Worker端

case driverStateChanged @ DriverStateChanged(driverId,exception) => {

//处理Driver State Changed

??handleDriverStateChanged(driverStateChanged)

}

给Master发消息

private[worker] def handleDriverStateChanged(driverStateChanged: DriverStateChanged): Unit = {

??val driverId = driverStateChanged.driverId

??val exception = driverStateChanged.exception

??val state = driverStateChanged.state

??state match {

? ? case DriverState.ERROR =>

? ?? ?logWarning(s"Driver?driverIdfailedwithunrecoverableexception:{exception.get}")

? ? case DriverState.FAILED =>

? ?? ?logWarning(s"Driver $driverId exited with failure")

? ? case DriverState.FINISHED =>

? ?? ?logInfo(s"Driver $driverId exited successfully")

? ? case DriverState.KILLED =>

? ?? ?logInfo(s"Driver $driverId was killed by user")

? ? case _ =>

? ?? ?logDebug(s"Driver?driverIdchangedstatetostate")

??}

//给master发送消息，告诉master，Driver状态发生变化了。

??sendToMaster(driverStateChanged)

Master端receive方法是负责接收Worker发消息的。根据Driver状态进行处理。

case DriverStateChanged(driverId,exception) => {

??state match {

? ? case DriverState.ERROR | DriverState.FINISHED | DriverState.KILLED | DriverState.FAILED =>

? ?? ?removeDriver(driverId,exception)

? ? case _ =>

? ?? ?throw new Exception(s"Received unexpected state update for driver?driverId:state")

??}

}

removeDriver方法：从自己的数据结构中remove掉

private def removeDriver(

? ? driverId: String,

? ? finalState: DriverState,

? ? exception: Option[Exception]) {

??drivers.find(d => d.id == driverId) match {

? ? case Some(driver) =>

? ?? ?logInfo(s"Removing driver: $driverId")

? ?? ?drivers -= driver

? ?? ?if (completedDrivers.size >= RETAINED_DRIVERS) {

? ?? ???val toRemove = math.max(RETAINED_DRIVERS / 10,1)

? ?? ???completedDrivers.trimStart(toRemove)

? ?? ?}

? ?? ?completedDrivers += driver

? ?? ???//删除持久化引擎，例如Zookeeper持久化数据。

? ?? ?persistenceEngine.removeDriver(driver)

? ?? ?driver.state = finalState

? ?? ?driver.exception = exception

? ?? ?driver.worker.foreach(w => w.removeDriver(driver))

? ?? ???//资源发生了变动，执行下schedule

? ?? ?schedule()

? ? case None =>

? ?? ?logWarning(s"Asked to remove unknown driver: $driverId")

??}

}

LaunchExecutor：先判断是否此时的路径是是activeMasterUrl

case LaunchExecutor(masterUrl,appId,execId,appDesc,cores_,memory_) =>

??if (masterUrl != activeMasterUrl) {

? ? logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor.")

创建Executor的工作目录

// Create the executor's working directory

val executorDir = new File(workDir,appId + "/" + execId)

启动ExecutorRunner

val manager = new ExecutorRunner(

??appId,

??execId,

??appDesc.copy(command = Worker.maybeUpdateSSLSettings(appDesc.command,

??cores_,

??memory_,

??self,

??workerId,

??host,

??webUi.boundPort,

??publicAddress,

??sparkHome,

??executorDir,

??workerUri,

??conf,

??appLocalDirs,ExecutorState.RUNNING)

executors(appId + "/" + execId) = manager

manager.start()

Start()方法通过fetchAndRunExecutor方法启动Executor

private[worker] def start() {

??workerThread = new Thread("ExecutorRunner for " + fullId) {

? ? override def run() { fetchAndRunExecutor() }

??}

fetchAndRunExecutor源码如下：

/**

* Download and run the executor described in our ApplicationDescription

*/

private def fetchAndRunExecutor() {

??try {

? ? // Launch the process

? ? val builder = CommandUtils.buildProcessBuilder(appDesc.command,new SecurityManager(conf),

? ?? ?memory,substituteVariables)

? ? val command = builder.command()

? ? val formattedCommand = command.asScala.mkString(""",""")

? ? logInfo(s"Launch command: $formattedCommand")

? ? builder.directory(executorDir)

? ? builder.environment.put("SPARK_EXECUTOR_DIRS",appLocalDirs.mkString(File.pathSeparator))

? ? // In case we are running this from within the Spark Shell,avoid creating a "scala"

? ? // parent process for the executor command

? ? builder.environment.put("SPARK_LAUNCH_WITH_SCALA","0")

? ? // Add webUI log urls

? ? val baseUrl =

? ?? ?s"http://publicAddress:webUiPort/logPage/?appId=appId&executorId=appId&executorId=execId&logType="

? ? builder.environment.put("SPARK_LOG_URL_STDERR",s"${baseUrl}stderr")

? ? builder.environment.put("SPARK_LOG_URL_STDOUT",s"${baseUrl}stdout")

? ? process = builder.start()

? ? val header = "Spark Executor Command: %sn%snn".format(

? ?? ?formattedCommand,"=" * 40)

? ? // Redirect its stdout and stderr to files

? ? val stdout = new File(executorDir,"stdout")

? ? stdoutAppender = FileAppender(process.getInputStream,stdout,conf)

? ? val stderr = new File(executorDir,"stderr")

? ? Files.write(header,UTF_8)

? ? stderrAppender = FileAppender(process.getErrorStream,conf)

? ? // Wait for it to exit; executor may exit with code 0 (when driver instructs it to shutdown)

? ? // or with nonzero exit code

? ? val exitCode = process.waitFor()

? ? state = ExecutorState.EXITED

? ? val message = "Command exited with code " + exitCode

//executor状态改变的时候给Worker发消息。

? ? worker.send(ExecutorStateChanged(appId,Some(message),Some(exitCode)))

Worker将消息发送给Master

sendToMaster(ExecutorStateChanged(appId,manager.state,None,None))

Master端处理的时候，还要给Driver发送消息

case ExecutorStateChanged(appId,message,exitStatus) => {

??val execOption = idToApp.get(appId).flatMap(app => app.executors.get(execId))

??execOption match {

? ? case Some(exec) => {

? ?? ?val appInfo = idToApp(appId)

? ?? ?val oldState = exec.state

? ?? ?exec.state = state

? ?? ?if (state == ExecutorState.RUNNING) {

? ?? ???assert(oldState == ExecutorState.LAUNCHING,

? ?? ?? ? s"executor?execIdstatetransferfromoldState to RUNNING is illegal")

? ?? ???appInfo.resetRetryCount()

? ?? ?}

//给Driver发送消息告诉Driver，Executor状态发生改变了。

? ?? ?exec.application.driver.send(ExecutorUpdated(execId,exitStatus))