spark-源码 master和worker启动

概述

Worker的启动都是通过启动shell脚本

Master启动

master启动从main函数开始,主要启动Rpc环境：RpcEnv（Rpc环境）：Akka和Netty

启动一个Master，通过启动 Shell 脚本start-master.sh

这个脚本实际启动 spark 的 master 类

start-master.sh -> spark-daemon.sh start org.apache.spark.deploy.master.Master

启动时会传入一些参数，比如cpu的执行核数，内存大小，app的main方法等

查看Master类的main方法

private[spark] object Master extends Logging {

val systemName = "sparkMaster"

private val actorName = "Master"

//master启动的入口，启动命令里会传入一些参数

def main(argStrings: Array[String]) {

SignalLogger.register(log)

//创建SparkConf val conf = new SparkConf

//保存参数到SparkConf

val args = new MasterArguments(argStrings, conf)

//创建ActorSystem

val (actorSystem, _, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort, conf)

//等待该主Actor结束

actorSystem.awaitTermination()

}

这里主要看startSystemAndActor方法

/**

* (1) 启动Master的actor system

* (2) 绑定端口

* (3) 启动webui和port

* (4) 启动rest服务和绑定端口

def startSystemAndActor(

host: String,

port: Int,

webUiPort: Int,

conf: SparkConf): (ActorSystem, Int, Int, Option[Int]) = {

val securityMgr = new SecurityManager(conf)

//利用AkkaUtils创建ActorSystem

val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf,

securityManager = securityMgr)

val actor = actorSystem.actorOf(

Props(classOf[Master], host, boundPort, webUiPort, securityMgr, conf), "Master")

....

}

spark底层通信是Akka

通过ActorSystem创建Actor -> actorSystem.actorOf, 就会执行Master的构造方法（也就是说上面调用actorOf方法的时候会创建actor，也就是调用Master的构造器）->然后执行Actor生命周期方法

执行Master的构造方法初始化一些变量

private[spark] class Master(

host: String,

port: Int,

webUiPort: Int,

val securityMgr: SecurityManager,

val conf: SparkConf)

extends Actor with ActorLogReceive with Logging with LeaderElectable {

//主构造器

//启用定期器功能

import context.dispatcher // to use Akka's scheduler.schedule()

val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)

def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss") // For application IDs

//woker超时时间

val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000

val RETAINED_APPLICATIONS = conf.getInt("spark.deploy.retainedApplications", 200)

val RETAINED_DRIVERS = conf.getInt("spark.deploy.retainedDrivers", 200)

val REAPER_ITERATIONS = conf.getInt("spark.dead.worker.persistence", 15)

val RECOVERY_MODE = conf.get("spark.deploy.recoveryMode", "NONE")

//一个HashSet用于保存WorkerInfo

val workers = new HashSet[WorkerInfo]

//一个HashMap用保存workid -> WorkerInfo

val idToWorker = new HashMap[String, WorkerInfo]

val addressToWorker = new HashMap[Address, WorkerInfo]

//一个HashSet用于保存客户端（SparkSubmit）提交的任务

val apps = new HashSet[ApplicationInfo]

//一个HashMap Appid-》 ApplicationInfo

val idToApp = new HashMap[String, ApplicationInfo]

val actorToApp = new HashMap[ActorRef, ApplicationInfo]

val addressToApp = new HashMap[Address, ApplicationInfo]

//等待调度的App

val waitingApps = new ArrayBuffer[ApplicationInfo]

val completedApps = new ArrayBuffer[ApplicationInfo]

var nextAppNumber = 0

val appIdToUI = new HashMap[String, SparkUI]

//保存DriverInfo

val drivers = new HashSet[DriverInfo]

val completedDrivers = new ArrayBuffer[DriverInfo]

val waitingDrivers = new ArrayBuffer[DriverInfo] // Drivers currently spooled for scheduling

主构造器执行完就会执行preStart –》执行完receive方法

//启动定时器，进行定时检查超时的worker

//重点看一下CheckForWorkerTimeOut

context.system.scheduler.schedule(0 millis, WORKER_TIMEOUT millis, self, CheckForWorkerTimeOut)

1、在第一次运行的时候需要等待多少时间；

　　2、循环的频率；

　　3、我们想发送消息的目标ActorRef ；

　　4、消息

preStart方法里创建了一个定时器，定时检查Woker的超时时间val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000默认为60秒

到此Master的初始化的主要过程到我们已经看到了，主要就是构造一个Master的Actor进行等待消息，并初始化了集合来保存task信息和Worker信息，和一个定时器来检查Worker的超时

Woker的启动

执行本地 shell 脚本salves.sh-> 通过读取配置文件，通过ssh的方式远程连接远端的worker节点，然后启动每个节点的 work 类

spark-daemon.sh start org.apache.spark.deploy.worker.Worker

脚本会启动org.apache.spark.deploy.worker.Worker 类

看Worker源码：

private[spark] object Worker extends Logging {

//Worker启动的入口

def main(argStrings: Array[String]) {

SignalLogger.register(log)

val conf = new SparkConf

val args = new WorkerArguments(argStrings, conf)

//新创ActorSystem和Actor

val (actorSystem, _) = startSystemAndActor(args.host, args.port, args.webUiPort, args.cores,

args.memory, args.masters, args.workDir)

actorSystem.awaitTermination()

}

这里最重要的是Woker的startSystemAndActor

def startSystemAndActor(

host: String,

port: Int,

webUiPort: Int,

cores: Int,

memory: Int,

masterUrls: Array[String],

workDir: String,

workerNumber: Option[Int] = None,

conf: SparkConf = new SparkConf): (ActorSystem, Int) = {

// The LocalSparkCluster runs multiple local sparkWorkerX actor systems

val systemName = "sparkWorker" + workerNumber.map(_.toString).getOrElse("")

val actorName = "Worker"

val securityMgr = new SecurityManager(conf)

//通过AkkaUtils ActorSystem

val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port,

conf = conf, securityManager = securityMgr)

val masterAkkaUrls = masterUrls.map(Master.toAkkaUrl(_, AkkaUtils.protocol(actorSystem)))

//通过actorSystem.actorOf创建Actor Worker-》执行构造器 -》 preStart -》 receice

actorSystem.actorOf(Props(classOf[Worker], host, boundPort, webUiPort, cores, memory,

masterAkkaUrls, systemName, "Worker", workDir, conf, securityMgr), name = "Worker")

(actorSystem, boundPort)

}

这里启动该Worker的Actor对象,到此Worker的启动初始化完成

Worker与Master通信

根据Actor生命周期接着Worker的preStart方法被调用，也就是说worker一起动就会给master发消息，进行注册（说白了就是把work信息存到master的一个list里）

override def preStart() {

assert(!registered)

createWorkDir()

context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])

shuffleService.startIfEnabled()

webUi = new WorkerWebUI(this, workDir, webUiPort)

webUi.bind()

//Worker向Master注册

registerWithMaster()

....

}

这里调用了一个registerWithMaster方法，开始向Master注册

def registerWithMaster() {

// DisassociatedEvent may be triggered multiple times, so don't attempt registration

// if there are outstanding registration attempts scheduled.

registrationRetryTimer match {

case None =>

registered = false

//开始注册

tryRegisterAllMasters()

....

}

registerWithMaster里通过匹配调用了tryRegisterAllMasters方法

，接下来看

private def tryRegisterAllMasters() {

//遍历master的地址

for (masterAkkaUrl <- masterAkkaUrls) {

logInfo("Connecting to master " + masterAkkaUrl + "...")

//Worker得到Mater actor的远程引用 val actor = context.actorSelection(masterAkkaUrl)

//向Master发送注册信息

actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)//Worker向Master发送了一个消息，注册内容包含,带去一些参数，id,主机，端口，cpu核数，内存等待 }

}

通过masterAkkaUrl和Master建立连接后

masterActor接受来自Worker的注册信息

override def receiveWithLogging = {

......

//接受来自Worker的注册信息

case RegisterWorker(id, workerHost, workerPort, cores, memory, workerUiPort, publicAddress) =>

{

logInfo("Registering worker %s:%d with %d cores, %s RAM".format(

workerHost, workerPort, cores, Utils.megabytesToString(memory)))

if (state == RecoveryState.STANDBY) {

// ignore, don't send response

//判断这个worker是否已经注册过

} else if (idToWorker.contains(id)) {

//如果注册过，告诉worker注册失败

sender ! RegisterWorkerFailed("Duplicate worker ID")

} else {

//没有注册过，把来自Worker的注册信息封装到WorkerInfo当中

val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,

sender, workerUiPort, publicAddress)

if (registerWorker(worker)) {

//用持久化引擎记录Worker的信息

persistenceEngine.addWorker(worker)

//向Worker反馈信息，告诉Worker注册成功

sender ! RegisteredWorker(masterUrl, masterWebUiUrl)

schedule()

} else {

val workerAddress = worker.actor.path.address

logWarning("Worker registration failed. Attempted to re-register worker at same " +

"address: " + workerAddress)

sender ! RegisterWorkerFailed("Attempted to re-register worker at same address: "

+ workerAddress)

}

注册成功后Worker向master发送心跳

override def receiveWithLogging = {

case RegisteredWorker(masterUrl, masterWebUiUrl) =>

logInfo("Successfully registered with master " + masterUrl)

registered = true

changeMaster(masterUrl, masterWebUiUrl)

//启动定时器，定时发送心跳Heartbeat

context.system.scheduler.schedule(0 millis, HEARTBEAT_MILLIS millis, self, SendHeartbeat)

if (CLEANUP_ENABLED) {

logInfo(s"Worker cleanup enabled; old application directories will be deleted in: $workDir")

context.system.scheduler.schedule(CLEANUP_INTERVAL_MILLIS millis,

CLEANUP_INTERVAL_MILLIS millis, self, WorkDirCleanup)

}

worker接受来自Master的注册成功的反馈信息,启动定时器，定时发送心跳Heartbeat

case SendHeartbeat =>

//worker发送心跳的目的就是为了报活

if (connected) { master ! Heartbeat(workerId) }

Master接收心跳消息，更新最后一次心跳时间

override def receiveWithLogging = {

....

case Heartbeat(workerId) => {

idToWorker.get(workerId) match {

case Some(workerInfo) =>

//更新最后一次心跳时间

workerInfo.lastHeartbeat = System.currentTimeMillis()

.....

}

记录并更新workerInfo.lastHeartbeat = System.currentTimeMillis()最后一次心跳时间

Master的定时任务会不断的发送一个CheckForWorkerTimeOut内部消息不断的轮询集合里的Worker信息，如果超过60秒就将Worker信息移除

//检查超时的Worker

case CheckForWorkerTimeOut => {

timeOutDeadWorkers()

}

timeOutDeadWorkers方法

def timeOutDeadWorkers() {

// Copy the workers into an array so we don't modify the hashset while iterating through it

val currentTime = System.currentTimeMillis()

val toRemove = workers.filter(_.lastHeartbeat < currentTime - WORKER_TIMEOUT).toArray

for (worker <- toRemove) {

if (worker.state != WorkerState.DEAD) {

logWarning("Removing %s because we got no heartbeat in %d seconds".format(

worker.id, WORKER_TIMEOUT/1000))

removeWorker(worker)

} else {

if (worker.lastHeartbeat < currentTime - ((REAPER_ITERATIONS + 1) * WORKER_TIMEOUT)) {

workers -= worker // we've seen this DEAD worker in the UI, etc. for long enough; cull it

}

如果（最后一次心跳时间<当前时间-超时时间）则判断为Worker超时，

将集合里的信息移除。

当下一次收到心跳信息时，如果是已注册过的，workerId不为空，但是WorkerInfo已被移除的条件，就会sender ! ReconnectWorker(masterUrl)发送一个重新注册的消息

case None =>

if (workers.map(_.id).contains(workerId)) {

logWarning(s"Got heartbeat from unregistered worker $workerId." +

" Asking it to re-register.")

//发送重新注册的消息

sender ! ReconnectWorker(masterUrl)

} else {

logWarning(s"Got heartbeat from unregistered worker $workerId." +

" This worker was never registered, so ignoring the heartbeat.")

}

Master与Worker启动的大致的通信流程到此ok