1.Client

Client中唯一的main方法如下：
def main(args: Array[String]) { if (!sys.props.contains("SPARK_SUBMIT")) { println("WARNING: This client is deprecated and will be removed in a future version of Spark") println("Use ./bin/spark-submit with "--master spark://host:port"") } val conf = new SparkConf() val driverArgs = new ClientArguments(args) if (!driverArgs.logLevel.isGreaterOrEqual(Level.WARN)) { conf.set("spark.akka.logLifecycleEvents", "true") } conf.set("spark.rpc.askTimeout", "10") conf.set("akka.loglevel", driverArgs.logLevel.toString.replace("WARN", "WARNING")) Logger.getRootLogger.setLevel(driverArgs.logLevel) //创建一个driverClient的Rpc环境，并将得到Master和client的远程引用 val rpcEnv = RpcEnv.create("driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf)) val masterEndpoints = driverArgs.masters.map(RpcAddress.fromSparkURL). map(rpcEnv.setupEndpointRef(Master.SYSTEM_NAME, _, Master.ENDPOINT_NAME)) //clientpoint rpcEnv.setupEndpoint("client", new ClientEndpoint(rpcEnv, driverArgs, masterEndpoints, conf)) //启动rpc环境 rpcEnv.awaitTermination() }

2.ClientEndpoint

ClientEndPoint可以看作给Driver传递消息的代理
属性简单，直接略过。
（1）构造函数为ClientEndPoint主构造函数
（2）onstart方法如下，
override def onStart(): Unit = { driverArgs.cmd match { case "launch" => //driver包装类，使得Worker和Driver的Rpc环境一样，做到共进退 val mainClass = "org.apache.spark.deploy.worker.DriverWrapper" //driver类路径 val classPathConf = "spark.driver.extraClassPath" val classPathEntries = sys.props.get(classPathConf).toSeq.flatMap { cp => cp.split(java.io.File.pathSeparator) } //driver库路径 val libraryPathConf = "spark.driver.extraLibraryPath" val libraryPathEntries = sys.props.get(libraryPathConf).toSeq.flatMap { cp => cp.split(java.io.File.pathSeparator) } //driver Jvm参数 val extraJavaOptsConf = "spark.driver.extraJavaOptions" val extraJavaOpts = sys.props.get(extraJavaOptsConf) .map(Utils.splitCommandString).getOrElse(Seq.empty) //将所有的在SparkConf中设置的属性赋值给java options的序列 val sparkJavaOpts = Utils.sparkJavaOpts(conf) //所有的javaOpts val javaOpts = sparkJavaOpts ++ extraJavaOpts val command = new Command(mainClass, Seq("{{WORKER_URL}}", "{{USER_JAR}}", driverArgs.mainClass) ++ driverArgs.driverOptions, sys.env, classPathEntries, libraryPathEntries, javaOpts) //将以上所有的信息封装在DriverDescription中 val driverDescription = new DriverDescription( driverArgs.jarUrl, driverArgs.memory, driverArgs.cores, driverArgs.supervise, command) //异步请求给master发送Driver的信息 ayncSendToMasterAndForwardReplySubmitDriverResponse case "kill" => val driverId = driverArgs.driverId ayncSendToMasterAndForwardReplyKillDriverResponse } }
（3）onstop方法简单，略过。
（4）receive方法如下，
override def receive: PartialFunction[Any, Unit] = { //收到master的响应回来的Driver信息，因为master是管家，Client是老板 case SubmitDriverResponse(master, success, driverId, message) => logInfo(message) if (success) { //将当前的activeMasterEndpoint设置为响应消息的master activeMasterEndpoint = master //找到driver的信息然后退出JVM pollAndReportStatus(driverId.get) } else if (!Utils.responseFromBackup(message)) { System.exit(-1) } case KillDriverResponse(master, driverId, success, message) => logInfo(message) if (success) { activeMasterEndpoint = master pollAndReportStatus(driverId),详见下① } else if (!Utils.responseFromBackup(message)) { System.exit(-1) } }
①pollAndReportStatus方法如下，用于找到driver的信息然后退出JVM
def pollAndReportStatus(driverId: String) { logInfo("... waiting before polling master for driver state") Thread.sleep(5000) logInfo("... polling master for driver state") //master请求得到Driver的信息 val statusResponse = activeMasterEndpoint.askWithRetryDriverStatusResponse statusResponse.found match { case false => logError(s"ERROR: Cluster master did not recognize $driverId") System.exit(-1) case true => logInfo(s"State of $driverId is ${statusResponse.state.get}") //返回的其实是worker的信息 (statusResponse.workerId, statusResponse.workerHostPort, statusResponse.state) match { case (Some(id), Some(hostPort), Some(DriverState.RUNNING)) => logInfo(s"Driver running on $hostPort ($id)") case _ => } statusResponse.exception.map { e => logError(s"Exception from cluster was: $e") e.printStackTrace() System.exit(-1) } System.exit(0) } }