畅聊Spark（五）内核解析

本文主要是介绍畅聊Spark（五）内核解析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

整体概念

           Apache Spark是一个开源的通用集群计算系统，提供了High-Level编程API，支持Scala、Java、Python三种编程语言，Spark内核是使用Scala语言编写的，通过基于Scala的函数式编程特性，在不同计算层面进行抽象。

计算抽象

Application

           用户编写Spark程序，完成一个计算任务的处理，由一个Driver程序和一组运行在Spark上的Executor组成。

Job

           用户程序中，每次调用Action时，逻辑上都会生成一个Job，一个Job包含多个Stage

Stage

           Stage包含两类，ShuffleMapStage和ResultStage，如果程序中调用了需要进行Shuffle计算的Operator，如groupByKey等，就会以Shuffle作为边界，分成ShuffleMapStage和ResultStage

TaskSet

           基于Stage可以直接银蛇为TaskSet，一个TaskSet封装了一次需要运算的、具有相同处理逻辑的Task，这些Task可以并行运算，粗粒度的调度是以TaskSet为单位。

Task

         Task是物理节点上运行的基本单位，Task包含两类，ShuffleMapTask和ResultTask，分别对应Stage中的ShuffleMapStage和ResultStage中的一个执行单元。

RPC通信架构

历史

           1.Spark早期版本采用Akka作为内部通信部件

           2.Spark 1.3引入Netty通信框架，为了解决Shuffle的大数据传输问题

           3.Spark 1.6中Akka和Netty可以配置使用，Netty完全实现了Akka在Spark中的功能。

           4.Spark 2.x中，抛弃了Akka，选择使用Netty

抛弃的Akka的原因

1.Akka不同版本之间无法通信，存在兼容性问题

           2.用户使用Akka和Spark中的Akka存在冲突

           3.Spark自身没有对Akka进行维护，需要新功能时，只能等待新版本，牵制了Spark的发展。

/*** A RpcEnv implementation must have a [[RpcEnvFactory]] implementation with an empty constructor* so that it can be created via Reflection.*/private[spark] object RpcEnv {     ……def create(name: String,bindAddress: String,advertiseAddress: String,port: Int,conf: SparkConf,securityManager: SecurityManager,clientMode: Boolean): RpcEnv = {val config = RpcEnvConfig(conf, name, bindAddress, advertiseAddress, port, securityManager,clientMode)new NettyRpcEnvFactory().create(config)}}

通信组件概览

           1.RPCEndpoint：RPC端点，Spark针对每一个节点（Client/Master/Worker）都称为一个PRC端点，而且都实现了RpcEndpoint接口，内部根据不同端点的需求，设计不同的消息和不同的业务处理，如果需要发送（询问）则调用Dispatcher。

         2.RpcEnv：RPC上下文环境，每个Rpc端点运行时依赖的上下文环境，称为RPCEnv。

           3.Dispatcher：消息分发器，针对RPC端点需要发送消息或者从远程RPC接收到的消息，分发至对应的指令收件箱/发件箱，如果指令接收方是自己存入收件箱，如果接收方非自身端点，则放入发件箱。

           4.Inbox：指令消息收件箱，一个本地端点对应一个收件箱，Dispatcher在每次Inbox存入消息时，都会将对应EndpointData加入内部，待Receiver Queue中，另外Dispatcher创建时，会启动一个单独线程进行轮询Receiver Queue，进行收件箱消费。

           5.OutBox：指令消息发件箱，一个远程端点对应一个发件箱，当消息放入Outbox后，接着将消息通过TransportClient发送出去，消息放入发件箱以及发送过程是在同一个线程中进行，这样做的主要原因是远程消息为RpcOutboxMessage，OneWayOutboxMessage两种消息，而针对需要应答的消息，直接发送且需要得到结果进行处理。

           6.TransportClient：Netty通信客户端，根据OutBox消息的receiver信息，请求对应远程TransportServer。

           7.TransportServer：Netty通信服务端，一个RPC端点一个TransportServer，接收远程消息后，调用Dispatcher分发消息至对应收件箱。

           注：TransportClient和TransportServer通信虚线，表示两个RPCEnv之间通信。

                        一个Outbox一个TransportClient

                        一个RPCEnv中存在两个RPCEndpoint，一个代表本身启动的PRC端点，另外一个为RPCEndpointVerifier。

内部实现

Master和Worker的通信（Standalone）

核心组件

核心交互流程

           橙色：提交用户Spark程序

           1.spark-submit脚本提交一个Spark程序，会创建一个ClientEndpoint对象，该对象负责和Master通信交互。

           2.ClientEndpoint向Master发送一个RequestSubmitDriver消息，表示提交用户程序。

           3.Master收到RequestSubmitDriver消息，向ClientEndpoint回复SubmitDriverResponse，表示用户程序已注册。

           4.ClientEndpoint向Master发送RequestDriverStatus消息，请求Driver状态。

           5.如果当前用户程序对应的Driver已经启动，则ClientEndpoint直接退出，完成提交用户程序。

           紫色：启动Driver进程

           1.Master内存维护者用户提交计算的任务Application，每次内存结构变更都会触发调度，向Worker发送LaunchDriver请求。

           2.Worker收到LaunchDriver消息，会启动一个DriverRunner线程去执行LaunchDriver的任务。

           3.DriverRunner线程在Worker上启动一个新的JVM实例，该JVM实例内运行一个Driver进程，该Driver会创建SparkContext对象。

           红色：注册Application

           1.创建SparkEnv对象，创建并管理一些基本组件。

           2.创建TaskScheduler，负责Task调度

           3.创建StandaloneSchedulerBackend，负责与ClusterManager进行资源协调。

           4.创建DirverEndpoint，其他组件可以与Driver进行通信。

           5.在StandaloneSchedulerBackend内部创建一个StandaloneAppClient，负责处理和Master的通信交互。

           6.StandaloneAppClient创建一个ClientEndpoint，实际负责与Master通信。

           7.ClientEndpoint向Master发送RegisterApplication消息，注册Application。

           8.Master收到RegisterApplication请求后，恢复ClientEndpoint一个RegisterApplication，表示已经注册成功。

           蓝色：启动Executor进程

           1.Master向Worker发送LaunchExecutor消息，请求启动Executor，同时Master会向Driver发送ExecutorAdded消息，表示Master新增了一个Executor（还未启动）。

           2.Worker收到LaunchExecutor消息，会启动一个ExecutorRunner线程去执行LaunchExecutor的任务。

           3.Worke向Master发送ExecutorStageChanged消息，通知Executor状态已发生变化。

           4.Master向Driver发送ExecutorUpdated消息，此时Executor已经启动。

           粉色：启动Executor进程

           1.StandaloneSchedulerBackend启动一个DriverEndpoint

           2.DriverEndpoint启动后，会周期性检查Driver维护的Executor的状态，如果有空闲的Executor则会调度任务执行。

           3.DriverEndpoint向TaskScheduler发送Resource Offer请求。

           4.如果有可用资源启动Task，则DriverEndpoint向Executor发送LaunchTask请求。

           5.Executor进程内部的CoarseGrainedExecutorBackend调用内部的Executor线程的LaunchTask方法启动Task。

           6.Executor线程内部维护一个线程池，创建一个TaskRunner线程并提交到线程池执行。

           绿色：Task运行完成

           1.Executor进程内部的Executor线程通知CoarseGrainedExecutorBackend，Task运行完成了。

           2. CoarseGrainedExecutorBackend向DriverEndpoint发送StatusUpdated消息，通知Driver运行的Task状态变更。

           3.StandaloneSchedulerBackend调用TaskScheduler的updateStatus方法更新Task状态。

4.随后StandaloneSchedulerBackend调用TaskScheduler的resourceOffers方法，调度其他任务运行。

整体应用

           1.Client运行时向Master发送启动驱动申请（发送RequestSubmitDriver指令）

           2.Master调度可用的Worker资源进行驱动安装（发送LaunchDriver指令）

           3.Worker运行DriverRunner进行驱动加载，并向Master发送应用注册请求（发送RegisterApplication指令）

           4.Master调度可用Worker资源进行应用的Executor安装（发送LaunchExecutor指令）

           5.Executor安装完毕后，向Driver注册驱动可用Executor资源（发送RegisterExecutor指令）

           6.最后是运行用户代码时，通过DAGScheduler，TaskScheduler封装为可以执行的TaskSetManager对象

           7.TaskSetManager对象与Driver中的Executor资源进行匹配，在队形的Executor中发布任务（发送LaunchTask指令）

           8.TaskRunner执行完毕后，调用DriverRunner提交给DAGScehduler，循环7直到任务完成。

SparkContext

           SparkContext是用户通往Spark集群中的唯一出口，任何需要使用Spark的地方都需要先创建SparkContext。

           SparkContext是在Driver程序里面启动的，可以看做Driver成和Spark集群的一个连接，SparkContext在初始化时，创建了很多对象。

           下图：列出了SparkContext在初始化创建时的一些主要组件的构建。

SparkContext结构和交互关系

           1.SparkContext是用户Spark执行任务上下文，用户程序内部使用Spark提供的Api直接或间接创建一个SparkContext。

           2.SparkEnv：用户执行的环境信息，包括通信相关的端点。

           3.RpcEnv：SparkContext中远程通信环境

           4.ApplicationDescription：应用程序描述信息，主要包含appName、maxCoes、memoryPerExecutorMB、coresPerExecutor、Command（CoarseGrainedExecutorBackend）、AppUiUrl等……

           5.ClientEndpoint：客户端端点，启动后向Master发起注册RegisterApplication请求。

           6.Master：接受RegisterApplication请求后，进行Worker资源分配，并向分配的资源发起LaunchExecutor指令。

           7.Worker：接受LaunchExecutor指令后，运行ExecutorRunner

           8.ExecutorRunner：运行applicationDescription的Command命令，最终Executor，同时向DriverEndpoint注册Executor信息。

MapReduce和Spark过程对比

存储子系统

           Storage模块主要分为两层：

           1.通信层：Storage模块采用的是master-slave结构来实现通信层，master和slave之间传输控制信息、状态信息，这些都是通过通信层来实现的。

           2.存储层：storage模块需要把数据存储到disk或memory上面，有可能还需要replicate到远端，这些都是由存储层来实现和提供相应接口的。

           而其他模块若是和Storage模块进行交互，Storage模块提供了一些统一的操作类BlockManager，外部类和storage模块打交道都需要通过调用BlockManager相应接口来实现。

           上图是Spark存储子系统中几个主要模块的关系图：

           1.CacheManager：RDD在进行计算的时候，通过CacheManager来获取数据，并通过CacheManager来存储计算结果。

           2.BlockManger：CacheManager在进行数据读取和存取的时候，主要是依赖BlockManager接口来操作，BlockManager决定数据是从内存（MemoryStore）还是从磁盘（DiskStore）获取。

           3.MemoryStore：负责将数据保存在内存或读取

           4.DiskStore：负责将数据写入磁盘或读取

           5.BlockManagerWorker：数据写入本地的MemoryStore或DiskStore是一个同步操作，为了容错还需要将数据负责到别的计算节点，以防止数据丢失时，还能恢复，数据复制的操作是异步完成，由BlockManagerWorker来处理这一部分事情。

         6.ConnectionManager：负责和其他计算节点建立连接，并负责数据的发送和接受。

           7.BlockManagerMaster：该模块只运行在Dirver Application所在的Executor，功能是负责记录下所有的BlockIds存储在哪个SlaveWorker上，比如RDD Task运行在机器A，所需要的是BlockId为3，但在机器A上没有BlockId为3的数值，这个时候Slave Worker需要通过BlockManager向BlockManagerMaster询问数据存储的位置，然后再通过ConnectionManager去获取。

Spark内存管理

           作为一个JVM进程，Executor的内存管理建立在JVM的内存管理之上，Spark对JVM的堆内（On-heap）空间，进行了更为详细的分配，以充分利用内存，同时Spark引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，进一步优化了内存的使用。

堆内内存

           堆内存储的大小，由Spark应用成启动时的-executor-memory或spark.executor.memory参数配置。

           Executor内运行的并发任务共享JVM堆内内存，这些任务在缓存RDD数据和在广播（Broadcast）数据时，占用的内存被规划为存储（Storage）内存，而任务在执行Shuffle时，占用的内存被规划为执行（Executor）内存，剩余的部分，不做特殊规划，那些Spark内部的对象实例，或者用户定义的Spark应用程序中的对象实例，均占用剩余的空间，不同的管理模式下，占用的空间也不尽相同。

堆外内存

           为了进一步优化内存的使用，及提高Shuffle时排序的效率，Spark引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，存储经过序列化的二进制数据，利用JDK Unsafe Api（从Spakr2.0开始，在管理堆外的存储内存时，不再基于Tachyon，而是和堆外的执行内存一样，基于JDK Unsafe API 实现），Spakr可以直接操作系统堆外内存，减少了不必要的内存开销，以及频繁的GC扫描和回收，提升了处理性能。

           堆外内存可以被精确的申请和释放，而且序列化的数据占用的的空间，可以被精确计算，所以相比堆内内存来说，降低了管理的难度，也降低了误差。

           默认情况下堆外内存是不启动的，可以通过配置spark.memory.offHeap.enabled参数启用，并由spark.memory.offHeap.size参数来设置堆外空间的消息。

           堆外内存和堆内内存的划分方式相同，所有运行中的并发任务共享存储内存和执行内存。

内存空间的分配

静态内存管理 – 堆内

静态内存管理 – 堆外

统一内存 – 堆内

统一内存 – 堆外

异常场景分析

Worker异常退出了

           1.Worker异常退出，比如说有意识的通过kill指令将Worker杀死。

           2.Worker在退出之前，会将自己管控的所有Executor给干掉。

           3.Worker需要定期向Master发送心跳消息，Worker进程挂了，心跳消息自然也没了，所以Master会在超时处理中得知。

           4.Master会把情况汇报给Driver

           5.Driver通过两方面却分配给自己的Executor挂了，一是Master发送来的消息，二是Driver没有在规定的时间内收到Executor的StatuUpdate，于是Driver会将注册的Executor移除。

           后果分析：

                      1.Worker异常退出，提交的Task无法正常的结束，会被再一次提交运行

                       2.如果所有的Worker都异常退出，则整个集群就不可用了。

                       3.需要有相应的程序来重启Worker进程，比如使用superisord或runit。

Executor异常退出了

           Executor作为Standalone集群部署方式下的，最底层单位。

           Executor异常退出，ExecutorRunner注意到异常，将情况通过ExecutorStateChanged汇报给Master。

           Master收到通知后，会要求Worker再次启动Executor。

           Worker收到LaunchExecutor指令，再次启动Executor。

Master异常退出了

           1.Worker没有汇报的对象了，也就是如果Executor再次跑飞了，Worker是不会将Executor启动起来的，因为没有Master的指令。

           2.无法向集群提交新任务。

           3.老的任务即便结束了，占用的资源也无法清除，因为资源清除的指令是由Master发出。

这篇关于畅聊Spark（五）内核解析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！