973
技術社區[雲棲]
spark源碼分析Master與Worker啟動流程篇
spark通信流程
概述
spark作為一套高效的分布式運算框架，但是想要更深入的學習它，就要通過分析spark的源碼，不但可以更好的幫助理解spark的工作過程，還可以提高對集群的排錯能力，本文主要關注的是Spark的Master的啟動流程與Worker啟動流程。
Master啟動
我們啟動一個Master是通過Shell命令啟動了一個腳本start-master.sh開始的，這個腳本的啟動流程如下
start-master.sh -> spark-daemon.sh start org.apache.spark.deploy.master.Master
我們可以看到腳本首先啟動了一個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和Actor
val (actorSystem, _, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort, conf)
//等待結束
actorSystem.awaitTermination()
}
這裏主要看startSystemAndActor方法
/**
* Start the Master and return a four tuple of:
* (1) The Master actor system
* (2) The bound port
* (3) The web UI bound port
* (4) The REST server bound port, if any
*/
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), actorName)
....
}
}
spark底層通信使用的是Akka
通過ActorSystem創建Actor -> actorSystem.actorOf, 就會執行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)
preStart方法裏創建了一個定時器，定時檢查Woker的超時時間 val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000 默認為60秒
到此Master的初始化的主要過程到我們已經看到了，主要就是構造一個Master的Actor進行等待消息，並初始化了一堆集合來保存Worker信息，和一個定時器來檢查Worker的超時
Master啟動時序圖
Woker的啟動
通過Shell腳本執行salves.sh -> 通過讀取slaves 通過ssh的方式啟動遠端的workerspark-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, actorName, workDir, conf, securityMgr), name = actorName)
(actorSystem, boundPort)
}
這裏Worker同樣的構造了一個屬於Worker的Actor對象,到此Worker的啟動初始化完成
Worker與Master通信
根據Actor生命周期接著Worker的preStart方法被調用
override def preStart() {
assert(!registered)
logInfo("Starting Spark worker %s:%d with %d cores, %s RAM".format(
host, port, cores, Utils.megabytesToString(memory)))
logInfo(s"Running Spark version ${org.apache.spark.SPARK_VERSION}")
logInfo("Spark home: " + sparkHome)
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建立連接
val actor = context.actorSelection(masterAkkaUrl)
//向Master發送注冊信息
actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)
}
}
通過masterAkkaUrl和Master建立連接後actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)Worker向Master發送了一個消息，帶去一些參數，id,主機，端口，cpu核數，內存等待
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)
}
}
}
這裏是最主要的內容；
receiveWithLogging裏會輪詢到Worker發送的消息，
Master收到消息後將參數封裝成WorkInfo對象添加到集合中，並加入到持久化引擎中sender ! RegisteredWorker(masterUrl, masterWebUiUrl)向Worker發送一個消息反饋
接下來看Worker的receiveWithLogging
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端的receiveWithLogging收到心跳消息
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.")
}
Worker與Master時序圖
Master與Worker啟動以後的大致的通信流程到此，接下來就是如何啟動集群上的Executor 進程計算任務了。
最後更新：2017-05-01 08:01:17