Spark基本架构及消息通信原理

比眉伴天荒 2021-11-05 11:32 602阅读 0赞

参考《图解Spark：核心技术与案例实战》

Spark 2.0版本

# 一、Spark基本架构 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70][]

**Application**：Spark应用程序  
**Driver**：Spark中的Driver即运行Application的main函数并且创建SparkContext，其中创建SparkContext为Spark的上下文。在Spark中由SparkContext负责与ClusterManager通信，进行资源申请、任务的分配和监控等；当Executor部分运行完成时候，Driver负责将SparkContext关闭。通常用SparkContext 代表Driver。  
**Cluster Manager**（集群管理器）：指在集群上获取资源的外部服务，目前有以下几种。

*  Standalone：Spark原生的资源管理，是由Master负责资源的管理。
 *  Hadoop Yarn：由Yarn中的ResourceManager负责资源的管理。
 *  Mesos：由Mesos中的Mesos Master负责资源管理
 *  Kubernetes（K8s）：Spark 2.3版本以上

**Worker**：集群中任何可以运行Application代码的节点，类似于YARN的NodeManager  
**Master**：Spark Standalone运行模型下的主节点，负责资源管理和分配资源来运行Spark Application。  
**Exeuctor**：Application运行Worker节点上的一个进程，该进程负责运行Task，并负责将数据存在内存或者磁盘上，每个Application都有各自一批的Executor。在Spark On Yarn 模式下，其进程名字为CoarseGraniedExecutorBackend，类似于Hadoop MapReduce中的Yarn Child。

# 二、Spark消息通信原理 #

**主要分成三个部分：整体框架；启动消息通信；运行时消息通信**

## 1、Spark通信整体框架 ##

**Spark2.0使用Netty作为master与worker的通信框架，Spark2.0之前使用的akka框架。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 1][]

![aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvOTQyNzk3LzIwMTcwOS85NDI3OTctMjAxNzA5MjExOTQ1NDI0OTMtNjg3ODM4MDgxLnBuZw][]

**RpcEndpoint**：Spark针对于每个节点（Client/Master/Worker）都称之一个RpcEndPoint，且都实现RpcEndpoint接口，内部根据不同端点的需求，设计不同的消息和不同的业务处理，如果需要发送（询问）则调用Dispatcher。一个RpcEndpoint经历的过程依次是：构建->onStart→receive→onStop。其中onStart在接收任务消息前调用，receive和receiveAndReply分别用来接收另一个RpcEndpoint（也可以是本身）send和ask过来的消息。

**RpcEndpointRef**：是对远程RpcEndpoint的一个引用。当我们需要**向一个具体的RpcEndpoint发送消息时，一般我们需要获取到该RpcEndpoint的引用，**然后通过该应用发送消息。

**RpcAddress**: 表示远程的RpcEndpointRef的地址，Host + Port。

**RpcEnv**：RPC上下文环境，**每个RpcEndPoint运行时依赖的上下文环境称之为RpcEnv。**RpcEnv负责RpcEndpoint整个生命周期的管理，包括：注册endpoint，endpoint之间消息的路由，以及停止endpoint。

**Dispatcher**：消息分发器，针对于RPC端点需要发送消息或者从远程RPC接收到的消息，分发至对应的指令收件箱/发件箱。

## 2、Spark启动消息通信 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlYmFvamluZw_size_16_color_FFFFFF_t_70][]

具体启动过程：

*  **当master启动之后再启动worker，worker会自动建立通信环境RpcEnv和EndPoint，**并且worker会向master发送注册信息RegisterWorker，master处理完毕后返回注册成功或者是失败的消息。HA条件下，worker可能需要注册到多个master上。在worker的tryRegisterAllMasters方法中创建注册线程池RegisterMasterThreadPool，把需要注册的请求放入这个线程池，通过此线程池启动注册线程。注册过程中，（注册线程）获取master EndPoint引用，（注册线程）调用RegisterWithMaster方法，根据master EndPoint引用的send方法给master发送注册消息RegisterWorker。
 *  **master接到注册消息RegisterWorker后，master会对worker发送的消息进行验证，记录。如果注册成功**，master发送registeredWorker给对应的worker，告诉worker已经完成注册，进行第3步骤，即worker定期发送心跳给master；**如果注册失败**，master会给worker发送registerWorkerFailed消息，worker会打印日志并结束worker的启动。master接到worker注册消息后，master先判断自己当前状态是否是STANDBY，如果是则忽略该消息；如果在注册列表发现了该worker编号，master则对worker发送注册失败的消息。判断完毕无误之后，调用registerWorker方法将此worker加入到master自己的列表，用于后续任务调度使用。
 *  **worker收到注册成功的消息后，会定时发送心跳给master，间隔时间是spark.，**方便master了解Worker状态。（注意timeout的四分之一才是心跳间隔哦）

## 3、Spark运行时消息通信 ##

**应用程序SparkContext向mater发送注册信息，并由master为该应用分配Exectour，exectour启动之后会向SparkContext发送注册成功消息，**然SparkContext的rdd触发Action之后会形成一个DAG，通过DAGScheduler进行划分Stage并将其转化成TaskSet，然后TaskScheduler向Executor发送执行消息，Executor接收到信息之后启动并且运行，最后是由Driver处理结果并回收资源。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 2][]

1.  启动SparkContext，启动SparkContext时会实例化一个Schedulerbackend对象，standalone模式下实际创建的是SparkDeploySchedulerbackend对象，在这个对象启动中会继承父类DriverEndpoint和ClientEndpoint这两个endpoint。ClientEndpoint的tryRegisterAllMasters方法中创建注册线程池RegisterMasterThreadPool，在该线程池中启动注册线程并向master发送registerApplication注册应用的消息。master接收到注册应用的消息，在registerApplication方法中记录应用信息，并将该应用加入等待运行应用列表，注册完毕后master发送成功消息RegisteredApplication给ClientEndPoint，同时调用startExecutorOnworker方法运行应用。
2.  ClientEndpoint接收到Master发送的RegisteredApplication消息，需要把注册标识registered置为true。
3.  在Master类的starExecutorsOnWorkers方法中分配资源运行应用程序时，调用allocateWorkerResourceToExecutors方法实现在Worker中启动Executor。当Worker收到Master发送过来的LaunchExecutor消息，先实例化ExecutorRunner对象，在ExecutorRunner启动中会创建进程生成器ProcessBuilder, 然后由该生成器使用command创建CoarseGrainedExecutorBackend对象，该对象是Executor运行的容器，最后Worker发送ExecutorStateChanged消息给Master，通知Executor容器【CoarseGrainedExecutorBackend】已经创建完毕。
4.  Master接收到Worker发送的ExecutorStateChanged消息
5.  在3中的CoarseGrainedExecutorBackend启动方法onStart中，会发送注册Executor消息RegisterExecutor给DriverEndpoint，DriverEndpoint先判断该Executor是否已经注册。
6.  CoarseGrainedExecutorBackend接收到Executor注册成功RegisteredExecutor消息时，在CoarseGrainedExecutorBackend容器中实例化Executor对象。启动完毕后，会定时向Driver发送心跳信息, 等待接收从DriverEndpoint发送执行任务的消息。
7.  CoarseGrainedExecutorBackend的Executor启动后接收从DriverEndpoint发送的LaunchTask执行任务消息，任务执行是在Executor的launchTask方法实现的。在执行时会创建TaskRunner进程，由该进程进行任务处理，处理完毕后发送StateUpdate消息返回给CoarseGrainedExecutorBackend。
8.  在TaskRunner执行任务完成时，会向DriverEndpoint发送StatusUpdate消息，DriverEndpoint接收到消息会调用TaskSchedulerImpl的statusUpdate方法，根据任务执行不同的结果处理，处理完毕后再给该Executor分配执行任务。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70]: /images/20211105/c0d87901b02349e18323ed38a2f60e0a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 1]: /images/20211105/6217242f3b9644eb89d26562652fc1ec.png
[aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvOTQyNzk3LzIwMTcwOS85NDI3OTctMjAxNzA5MjExOTQ1NDI0OTMtNjg3ODM4MDgxLnBuZw]: /images/20211105/fd21a3224d6f4bc0a184e6b13c5bb797.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlYmFvamluZw_size_16_color_FFFFFF_t_70]: /images/20211105/ba527c9c2e2e41f0bc15877d6b8199e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70 2]: /images/20211105/621c027f065b4c58b3743591b23b4f79.png