Spark源码分析之Master的启动流程
准备
本文主要对Master的启动流程源码进行分析。Spark源码版本为2.3.1。
阅读源码首先从启动脚本入手，看看首先加载的是哪个类，我们看一下start-master.sh启动脚本中的具体内容。
脚本代码
可以看到这里加载的类是org.apache.spark.deploy.master.Master，好那我们的源码寻觅之旅就从这开始...
源码分析
打开源码，我们发现Master是伴生关系的一组类，我们直接定位到Master的main函数
//主方法
def main(argStrings: Array[String]) {
Thread.setDefaultUncaughtExceptionHandler(new SparkUncaughtExceptionHandler(
exitOnUncaughtException = false))
Utils.initDaemon(log)
val conf = new SparkConf
val args = new MasterArguments(argStrings, conf)
/**
* 创建RPC 环境和Endpoint （RPC 远程过程调用）,在Spark中 Driver， Master ，Worker角色都有各自的Endpoint，相当于各自的通信邮箱。
*
*/
val (rpcEnv, _, _) = startRpcEnvAndEndpoint(args.host, args.port, args.webUiPort, conf)
rpcEnv.awaitTermination()
}
发现该函数除了做了一些配置文件和args参数的准备之外，调用了startRpcEnvAndEndpoint函数，我们跟进去看看
/**
* Start the Master and return a three tuple of:
* (1) The Master RpcEnv
* (2) The web UI bound port
* (3) The REST server bound port, if any
*/
def startRpcEnvAndEndpoint(
host: String,
port: Int,
webUiPort: Int,
conf: SparkConf): (RpcEnv, Int, Option[Int]) = {
val securityMgr = new SecurityManager(conf)
/**
* 创建RPC(Remote Procedure Call )环境 ,Remote Procedure Call
* 这里只是创建准备好Rpc的环境，后面会向RpcEnv中注册 角色【Driver,Master,Worker,Executor】
*/
val rpcEnv = RpcEnv.create(SYSTEM_NAME, host, port, conf, securityMgr)
.....
}
首先上面这段代码通过RpcEnv 构建了一个RPC通信环境，为之后的RPC通信做准备，Spark底层的通信是基于Netty的NIO模型，
这里每个Rpc端点运行时依赖的上下文环境称之为RpcEnv。
接下来我们直接跟踪到create方法中
def create(
name: String,
bindAddress: String,
advertiseAddress: String,
port: Int,
conf: SparkConf,
securityManager: SecurityManager,
numUsableCores: Int,
clientMode: Boolean): RpcEnv = {
val config = RpcEnvConfig(conf, name, bindAddress, advertiseAddress, port, securityManager,
numUsableCores, clientMode)
//创建NettyRpc 环境
new NettyRpcEnvFactory().create(config)
}
可以看到，这里构建了NettyRpcEnvFactory来创建NettyRpc 环境,NettyRpcEnvFactory的create函数又做了那些事呢？
def create(config: RpcEnvConfig): RpcEnv = {
val sparkConf = config.conf
// Use JavaSerializerInstance in multiple threads is safe. However, if we plan to support
// KryoSerializer in future, we have to use ThreadLocal to store SerializerInstance
val javaSerializerInstance =
new JavaSerializer(sparkConf).newInstance().asInstanceOf[JavaSerializerInstance]
/**
* 创建nettyRPC通信环境。
* 当new NettyRpcEnv时会做一些初始化：
* Dispatcher：这个对象中有存放消息的队列和消息的转发
* TransportContext：可以创建了NettyRpcHandler
*/
val nettyEnv =
new NettyRpcEnv(sparkConf, javaSerializerInstance, config.advertiseAddress,
config.securityManager, config.numUsableCores)
if (!config.clientMode) {
//启动nettyRPCEnv
val startNettyRpcEnv: Int => (NettyRpcEnv, Int) = { actualPort =>
nettyEnv.startServer(config.bindAddress, actualPort)
(nettyEnv, nettyEnv.address.port)
}
try {
//以上 startNettyRpcEnv 匿名函数在此处会最终被调用，当匿名函数被调用时，重点方法是483行 nettyEnv.startServer 方法
Utils.startServiceOnPort(config.port, startNettyRpcEnv, sparkConf, config.name)._1
} catch {
case NonFatal(e) =>
nettyEnv.shutdown()
throw e
}
}
nettyEnv
}
可以看到上面首先构建了一个序列化的实例对象，然后开始着手构架nettyRPC通信环境。在new NettyRpcEnv（）时会做一些初始化的工作，如下所示
/**
* dispatcher 这个对象中有消息队列和消息的循环获取转发
*/
private val dispatcher: Dispatcher = new Dispatcher(this, numUsableCores)
/**
* TransportContext 中会创建 NettyPpcHandler
* TransportContext 这个对象中参数类型 RpcHandler 就是这里的 NettyRpcHandler
*/
private val transportContext = new TransportContext(transportConf,
new NettyRpcHandler(dispatcher, this, streamManager))
这里的Dispatcher是一个消息分发器，针对于RPC端点需要发送消息或者从远程RPC接收到的消息，分发至对应的指令收件箱/发件箱。
这里的transportContext 是构建传输的上下文环境，用于创建TransportServer和TransportClientFactory同时，通过TransportChannelHandler来设置Netty的Channel pipelines。
下面我们回到NettyRpcEnvFactory的create方法
val nettyEnv =
new NettyRpcEnv(sparkConf, javaSerializerInstance, config.advertiseAddress,
config.securityManager, config.numUsableCores)
if (!config.clientMode) {
//启动nettyRPCEnv
val startNettyRpcEnv: Int => (NettyRpcEnv, Int) = { actualPort =>
nettyEnv.startServer(config.bindAddress, actualPort)
(nettyEnv, nettyEnv.address.port)
}
try {
//以上 startNettyRpcEnv 匿名函数在此处会最终被调用，当匿名函数被调用时，重点方法是483行 nettyEnv.startServer 方法
Utils.startServiceOnPort(config.port, startNettyRpcEnv, sparkConf, config.name)._1
} catch {
case NonFatal(e) =>
nettyEnv.shutdown()
throw e
}
}
nettyEnv 准备好了之后，会构建一个函数startNettyRpcEnv，然后startServiceOnPort在会调用startNettyRpcEnv，进而调用nettyEnv的startServer函数，来启动Server
//启动Rpc 服务
def startServer(bindAddress: String, port: Int): Unit = {
val bootstraps: java.util.List[TransportServerBootstrap] =
if (securityManager.isAuthenticationEnabled()) {
java.util.Arrays.asList(new AuthServerBootstrap(transportConf, securityManager))
} else {
java.util.Collections.emptyList()
}
/**
* transportContext已经被创建，这里createServer 就会绑定地址和端口，启动Netty Rpc 服务
*/
server = transportContext.createServer(bindAddress, port, bootstraps)
dispatcher.registerRpcEndpoint(
RpcEndpointVerifier.NAME, new RpcEndpointVerifier(this, dispatcher))
}
由于直接初始化时transportContext就已经被创建，这里createServer 就会绑定地址和端口，启动Netty Rpc 服务，createServer最在构建TransportServer实例时，会调用init初始化方法，在这里以前了解过Netty的同学就会非常熟悉了.
private void init(String hostToBind, int portToBind) {
IOMode ioMode = IOMode.valueOf(conf.ioMode());
EventLoopGroup bossGroup =
NettyUtils.createEventLoop(ioMode, conf.serverThreads(), conf.getModuleName() + "-server");
EventLoopGroup workerGroup = bossGroup;
PooledByteBufAllocator allocator = NettyUtils.createPooledByteBufAllocator(
conf.preferDirectBufs(), true /* allowCache */, conf.serverThreads());
bootstrap = new ServerBootstrap()
.group(bossGroup, workerGroup)
.channel(NettyUtils.getServerChannelClass(ioMode))
.option(ChannelOption.ALLOCATOR, allocator)
.childOption(ChannelOption.ALLOCATOR, allocator);
this.metrics = new NettyMemoryMetrics(
allocator, conf.getModuleName() + "-server", conf);
if (conf.backLog() > 0) {
bootstrap.option(ChannelOption.SO_BACKLOG, conf.backLog());
}
if (conf.receiveBuf() > 0) {
bootstrap.childOption(ChannelOption.SO_RCVBUF, conf.receiveBuf());
}
if (conf.sendBuf() > 0) {
bootstrap.childOption(ChannelOption.SO_SNDBUF, conf.sendBuf());
}
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
RpcHandler rpcHandler = appRpcHandler;
for (TransportServerBootstrap bootstrap : bootstraps) {
rpcHandler = bootstrap.doBootstrap(ch, rpcHandler);
}
//初始化网络通信管道
context.initializePipeline(ch, rpcHandler);
}
});
InetSocketAddress address = hostToBind == null ?
new InetSocketAddress(portToBind): new InetSocketAddress(hostToBind, portToBind);
channelFuture = bootstrap.bind(address);
channelFuture.syncUninterruptibly();
port = ((InetSocketAddress) channelFuture.channel().localAddress()).getPort();
logger.debug("Shuffle server started on port: {}", port);
}
init 方法就是去初始化一个绑定的主机和端口，创建nettyRPC通信，通过之前的rpcHandler来初始化网络通信管道，同时会调用createChannelHandler函数，创建处理消息的 channelHandler用于处理客户端请求消息和服务端回应消息
private TransportChannelHandler createChannelHandler(Channel channel, RpcHandler rpcHandler) {
TransportResponseHandler responseHandler = new TransportResponseHandler(channel);
TransportClient client = new TransportClient(channel, responseHandler);
TransportRequestHandler requestHandler = new TransportRequestHandler(channel, client,
rpcHandler, conf.maxChunksBeingTransferred());
/**
* 由以上 responseHandler client requestHandler 三个handler构建 TransportChannelHandler
* new TransportChannelHandler 这个对象中有 【channelRead() 方法】,用于读取接收到的消息
*/
return new TransportChannelHandler(client, responseHandler, requestHandler,
conf.connectionTimeoutMs(), closeIdleConnections);
}
最终Rpc的环境就准备好了，后面会向RpcEnv中注册 角色 Driver,Master,Worker,Executor。我们回到Master的startRpcEnvAndEndpoint函数
/**
* 向RpcEnv 中 注册Master
*
* rpcEnv.setupEndpoint(name,new Master)
* 这里new Master 的Master 是一个伴生类，继承了 ThreadSafeRpcEndpoint，归根结底继承到了 Trait 接口 RpcEndpoint
val masterEndpoint: RpcEndpointRef = rpcEnv.setupEndpoint(ENDPOINT_NAME,
new Master(rpcEnv, rpcEnv.address, webUiPort, securityMgr, conf))
val portsResponse = masterEndpoint.askSync[BoundPortsResponse](BoundPortsRequest)
(rpcEnv, portsResponse.webUIPort, portsResponse.restPort)
这里会向RpcEnv中注册Master，调用了setupEndpoint函数，传入了Master的实例对象，这里的Master 是一个伴生类，继承了 ThreadSafeRpcEndpoint，也是RpcEndpoint的实现类，而什么是RpcEndpoint？
RpcEndpoint：RPC端点 ，Spark针对于每个节点（Client/Master/Worker）都称之一个Rpc端点 ,且都实现RpcEndpoint接口，内部根据不同端点的需求，设计不同的消息和不同的业务处理，如果需要发送（询问）则调用Dispatcher。
EndPoint中存在
- onstart() :启动当前Endpoint
- receive() :负责收消息
- receiveAndReply():接受消息并回复
同时Endpoint 还有各自的引用，方便其他Endpoint发送消息，直接引用对方的EndpointRef 即可找到对方的Endpoint，上面源码中的masterEndpoint 就是Master的Endpoint引用 RpcEndpointRef 。
RpcEndpointRef中存在
- send():发送消息
- ask() :请求消息，并等待回应。
接下来我们看看NettyRpcEnv中的setupEndpoint函数
override def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef = {
dispatcher.registerRpcEndpoint(name, endpoint)
}
上面直接调用了dispatcher实例，注册RpcEndpoint
def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {
val addr = RpcEndpointAddress(nettyEnv.address, name)
val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)
synchronized {
if (stopped) {
throw new IllegalStateException("RpcEnv has been stopped")
}
//这里 new EndpointData使用到了endpoint，当new Inbox 时向消息队列中放入OnStart样例类标识
if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {
throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")
}
//获取刚刚封装的EndPointData
val data: EndpointData = endpoints.get(name)
endpointRefs.put(data.endpoint, data.ref)
/**
* receivers 这个消息队列中放着应该去哪个Endpoint 中获取Message 处理
* 这里其实就是进入 Dispatcher 当前这个类中的 MessageLoop 方法。这个方法当new Dispatcher后会一直运行。
* 将消息放入待处理的消息队列中，消息首先找到对应的Endpoint ,再会获取当前Endpoint的Inbox 中message,使用process 方法处理
*/
receivers.offer(data) // for the OnStart message
}
endpointRef
}
上面的new EndpointData使用到了endpoint，EndpointData其实就是对endpoint和endpointRef的封装，同时内部还构建了Inbox
private class EndpointData(
val name: String,
val endpoint: RpcEndpoint,
val ref: NettyRpcEndpointRef) {
//将endpoint封装到Inbox中
val inbox = new Inbox(ref, endpoint)
}
这里的Inbox是其实就是消息收件箱，一个本地端点对应一个收件箱，Dispatcher在每次向Inbox存入消息时，都将对应EndpointData加入内部待Receiver Queue中，另外Dispatcher创建时会启动一个单独线程进行轮询Receiver Queue，进行收件箱消息消费。
其实Inbox实例在构建过程中也会有消息的存入。
private[netty] class Inbox(
val endpointRef: NettyRpcEndpointRef,
val endpoint: RpcEndpoint)
extends Logging {
....
@GuardedBy("this")
protected val messages = new java.util.LinkedList[InboxMessage]()
// OnStart should be the first message to process
//当注册endpoint时都会调用这个异步方法，messags中放入一个OnStart样例类消息对象
inbox.synchronized {
messages.add(OnStart)
}
....
可以看到，构建inbox时会向messages中加入一个OnStart消息，该消息会被inbox类中的process所消费，只不过触发时机还在后面。
回到Dispatcher的registerRpcEndpoint函数
if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {
throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")
}
//获取刚刚封装的EndPointData
val data: EndpointData = endpoints.get(name)
endpointRefs.put(data.endpoint, data.ref)
/**
* receivers 这个消息队列中放着应该去哪个Endpoint 中获取Message 处理
* 这里其实就是进入 Dispatcher 当前这个类中的 MessageLoop 方法。这个方法当new Dispatcher后会一直运行。
* 将消息放入待处理的消息队列中，消息首先找到对应的Endpoint ,再会获取当前Endpoint的Inbox 中message,使用process 方法处理
*/
receivers.offer(data) // for the OnStart message
}
endpointRef
这里发现刚才构建的EndpointData会被放入到endpoints中，endpoints和endpointRefs的类型都是ConcurrentHashMap，对这两个容器进行相应的操作之后，就会调用receivers.offer(data)。
receivers是Dispatcher中的一个消息队列，这个消息队列还有一个线程池来进行消息的消费，整个模式构成一个生产者消费者模式，因此这里将data消息加入到消息队列中，会触发线程消费。
private val threadpool: ThreadPoolExecutor = {
val availableCores =
if (numUsableCores > 0) numUsableCores else Runtime.getRuntime.availableProcessors()
val numThreads = nettyEnv.conf.getInt("spark.rpc.netty.dispatcher.numThreads",
math.max(2, availableCores))
val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, "dispatcher-event-loop")
for (i <- 0 until numThreads) {
pool.execute(new MessageLoop)
}
pool
}
/** Message loop used for dispatching messages. */
private class MessageLoop extends Runnable {
override def run(): Unit = {
try {
while (true) {
try {
//take 出来消息一直处理
val data: EndpointData = receivers.take()
if (data == PoisonPill) {
// Put PoisonPill back so that other MessageLoops can see it.
receivers.offer(PoisonPill)
return
}
//调用process 方法处理消息
data.inbox.process(Dispatcher.this)
} catch {
case NonFatal(e) => logError(e.getMessage, e)
}
}
} catch {
case ie: InterruptedException => // exit
}
}
}
可以看到上面的MessageLoop中，通过消息队列receivers取出一条消息，该消息的类型为EndpointData，内部都封装了inbox实例，因此直接调用该实例的process函数去处理消息，我们之前在inbox实例构建过程中，发现会向inbox内部的消息容器中放入一条onStart消息，因此我们看一下process函数是如何处理该消息的。
def process(dispatcher: Dispatcher): Unit = {
case OnStart =>
//调用Endpoint 的onStart函数
endpoint.onStart()
if (!endpoint.isInstanceOf[ThreadSafeRpcEndpoint]) {
inbox.synchronized {
if (!stopped) {
enableConcurrent = true
}
}
}
}
process中对onStart的处理为，调用endpoint的onStart函数，而endpoint我们还记得是啥吗，这里是我们直接构建的Master实例，因此我们回到Master中去看看onStart()函数的处理过程。
override def onStart(): Unit = {
.....
val (persistenceEngine_, leaderElectionAgent_) = RECOVERY_MODE match {
case "ZOOKEEPER" =>
logInfo("Persisting recovery state to ZooKeeper")
val zkFactory =
new ZooKeeperRecoveryModeFactory(conf, serializer)
(zkFactory.createPersistenceEngine(), zkFactory.createLeaderElectionAgent(this))
case "FILESYSTEM" =>
val fsFactory =
new FileSystemRecoveryModeFactory(conf, serializer)
(fsFactory.createPersistenceEngine(), fsFactory.createLeaderElectionAgent(this))
case "CUSTOM" =>
val clazz = Utils.classForName(conf.get("spark.deploy.recoveryMode.factory"))
val factory = clazz.getConstructor(classOf[SparkConf], classOf[Serializer])
.newInstance(conf, serializer)
.asInstanceOf[StandaloneRecoveryModeFactory]
(factory.createPersistenceEngine(), factory.createLeaderElectionAgent(this))
case _ =>
(new BlackHolePersistenceEngine(), new MonarchyLeaderAgent(this))
}
persistenceEngine = persistenceEngine_
leaderElectionAgent = leaderElectionAgent_
}
onStart函数涉及到webUI的启动，applicationMetricsSystem和masterMetricsSystem的启动过程，如果设置了restServer还涉及到启动过程。
同时Master的onStart函数的部分代码如上所示，我么可以看到，这里涉及到如何在HA的环境下选主的过程。内部会根据配置决定采用哪种方式，是ZOOKEEPER还是文件系统的方式来进行。
生产环境下一般采用Zookeeper做HA，Zookeeper会自动化管理 Master的切换；
采用Zookeeper做HA的时候，Zookeeper会负责保存整个Spark集群运行时候的元数据：workers、Drivers、Applications、Executors；
Zookeeper遇到当前Active级别的Master出现故障的时候会从StandbyMaster中选取一台作为Active Master，但是要注意，被选举后到成为真正的ActiveMaster之间需要从Zookeeper中获取集群当前运行状态的元数据信息并进行恢复；
在Master切换的过程中，所有的已经在运行的程序皆正常运行！因为Spark Application在运行前就已经通过ClusterManager获得了计算资源，所以在运行时Job本身的调度和处理和Master是没有任何关系的！
在Master的切换过程中唯一的影响是不能提交新的job：一方面不能提交新的应用程序给集群，因为只有ActiveMaster才能接受新的程序的提交请求；另外一方面，已经运行的程序中也不能够因为Action操作触发新的Job的提交请求；
