Spark架构解析(转)

Application:

Application是建立了SparkContext实例对象的Spark用户，包含了Driver程序, 
Spark-shell是一个应用程序，由于spark-shell在启动的时候建立了一个SparkContext对象，其名称为sc,也就是说只要建立一个SparkContext就有对应的application，而一个action对应一个driver.相对应的transformation只产生元数据。

Job:

和action相对应，每个action例如count、saveAsTextFile等都会对应一个job实例，该job实例包含多任务的并行计算。

Stage：

一个job会被拆分红不少任务，每一组任务被称为stage，这个MapReduce的map和reduce任务很像，划分stage的依据在于：stage开始通常是因为读取外部数据或者shuffle数据、一个stage的结束通常是因为发生shuffle（例如reduceByKey操做）或者整个job结束时，例如要把数据放到hdfs等存储系统上

Driver Program：

运行main函数而且建立SparkContext实例的程序

Cluster Manager：

在集群上获取资源的外部服务 （如：standalone,Mesos,yarn）。集群资源的管理外部服务，在spark上如今有standalone、yarn、mesos等三种集群资源管理器，spark自带的standalone模式可以知足大部分的spark计算环境对集群资源管理的需求，基本上只有在集群中运行多套计算框架的时候才考虑yarn和mesos

Worker Node：

集群中能够运行应用代码的工做节点，至关于Hadoop的slave节点

Executor：

在一个Worker Node上为应用启动的工做进程，在进程中赋值任务的运行，而且负责将数据存放在内存或磁盘上，必须注意的是，每一个应用在一个Worker Node上只会有一个Executor，在Executor内部经过多线程的方式并发处理应用的任务。每一个应用对应独立的executor.运行在不一样的jvm.

Task：

被Driver送到Executor上的工做单元，一般状况下一个task会处理一个split的数据，每一个split通常就是一个Block块的大小。split即内存槽

stage:

一个job会被拆分红多组任务，每组任务叫作stage(就像mapreduce分为map任务和reduce任务)。通常stage起始于从外部读数据，结束于shuffle阶段，或任务结束。

spark架构

 
任何Spark的进程都是一个JVM进程，既然是一个JVM进程，那么就能够配置它的堆大小（-Xmx和-Xms）,可是进程怎么使用堆内存和为何须要它呢？下面是一个JVM堆空间下Spark的内存分配状况 
 
默认状况下，Spark进程的堆空间是512mb，为了安全考虑同时避免OOM,Spark只容许利用90%的堆空间，spark中使用spark.storage.safetyFraction用来配置该值（默认是0.9). Spark做为一个内存计算工具，Spark能够在内存中存储数据，它只是把内存做为他的LRU缓存，这样大量的内存被用来缓存正在计算的数据，该部分占safe堆的60%，Spark使用spark.storage.memoryFraction控制该值，若是想知道Spark中能缓存了多少数据，能够统计全部Executor的堆大小，乘上safeFraction和memoryFraction，默认是54%,这就是Spark可用缓存数据使用的堆大小。 
该部分介绍shuffle的内存使用状况，它经过 堆大小 * spark.shuffle.safetyFraction * spark.shuffle.memoryFraction。 spark.shuffle.safetyFraction的默认值是0.8， spark.shuffle.memoryFraction的默认值是0.2，因此最终只能最多使堆空间的16%用于shuffle。，可是一般spark会使用这块内存用于shuffle中一些别的任务，当执行shuffle时，有时对数据进行排序，当进行排序时，须要缓冲排完序后的数据（注意不能改变LRU缓冲中的数据，由于后面可能要重用），这样就须要大量的RAM存储排完序后的数据块，当没有足够的内存用于排序，参考外排的实现，能够一块一块的排序，而后最终合并。 
最后要讲到的一块内存是”unroll”,该快内存用于unroll计算以下：spark.storage.unrollFraction * spark.storage.memoryFraction * spark.storage.safetyFraction 。当咱们须要在内存展开数据块的时候使用，那么为何须要展开呢？由于spark容许以序列化和非序列化两种方式存储数据，序列化后的数据没法直接使用，因此使用时必需要展开。该部份内存占用缓存的内存，因此若是须要内存用于展开数据时，若是这个时候内存不够，那么Spark LRU缓存中的数据会删除一些快。 
此时应该清楚知道spark怎么使用JVM中堆内存了，如今切换到集群模式，当你启动一个spark集群，如何看待它，下图是YARN模式下的 
 
当运行在yarn集群上时，Yarn的ResourceMananger用来管理集群资源，集群上每一个节点上的NodeManager用来管控所在节点的资源，从yarn的角度来看，每一个节点看作可分配的资源池，当向ResourceManager请求资源时，它返回一些NodeManager信息，这些NodeManager将会提供execution container给你，每一个execution container就是知足请求的堆大小的JVM进程，JVM进程的位置是由ResourceMananger管理的，不能本身控制，若是一个节点有64GB的内存被yarn管理（经过yarn.nodemanager.resource.memory-mb配置),当请求10个4G内存的executors时，这些executors可能运行在同一个节点上。 
当在yarn上启动spark集群上，能够指定executors的数量（-num-executors或者spark.executor.instances)，能够指定每一个executor使用的内存（-executor-memory或者spark.executor.memory),能够指定每一个executor使用的cpu核数（-executor-cores或者spark.executor.cores),指定每一个task执行使用的core数（spark.task.cpus),也能够指定driver应用使用的内存（-driver-memory和spark.driver.memory) 
当在集群上执行应用时，job会被切分红stages,每一个stage切分红task,每一个task单独调度，能够把executor的jvm进程看作task执行池，每一个executor有 spark.executor.cores / spark.task.cpus execution 个执行槽，这里有个例子：集群有12个节点运行Yarn的NodeManager，每一个节点有64G内存和32的cpu核，每一个节点能够启动2个executor，每一个executor的使用26G内存，剩下的内用系统和别的服务使用，每一个executor有12个cpu核用于执行task,这样整个集群有12 machines * 2 executors per machine * 12 cores per executor / 1 core = 288 个task执行槽，这意味着spark集群能够同时跑288个task,整个集群用户缓存数据的内存有0.9 spark.storage.safetyFraction * 0.6 spark.storage.memoryFraction * 12 machines * 2 executors per machine * 26 GB per executor = 336.96 GB. 
到目前为止，咱们已经了解了spark怎么使用JVM的内存以及集群上执行槽是什么，目前为止尚未谈到task的一些细节，这将在另外一个文章中提升，基本上就是spark的一个工做单元，做为exector的jvm进程中的一个线程执行，这也是为何spark的job启动时间快的缘由，在jvm中启动一个线程比启动一个单独的jvm进程块（在hadoop中执行mapreduce应用会启动多个jvm进程） 
下面将关注spark的另外一个抽象：partition, spark处理的全部数据都会切分红partion,一个parition是什么以及怎么肯定，partition的大小彻底依赖数据源，spark中大部分用于读取数据的方法均可以指定生成的RDD中partition的个数，当从hdfs上读取一个文件时，会使用Hadoop的InputFormat来处理，默认状况下InputFormat返回每一个InputSplit都会映射RDD中的一个Partition,大部分存储在HDFS上的文件每一个数据块会生成一个InputSplit,每一个数据块大小为64mb和128mb,由于HDFS上面的数据的块边界是按字节来算的（64mb一个块），可是当数据被处理是，它又要按记录进行切分，对于文本文件来讲切分的字符就是换行符，对于sequence文件来讲，他是块结束，若是是压缩文件，整个文件都被压缩了，它不能按行进行切分了，整个文件只有一个inputsplit,这样spark中也会只有一个parition,在处理的时候须要手动的repatition。

Spark架构与做业执行流程简介

Client：客户端进程，负责提交做业到Master。 
Master：Standalone模式中主控节点，负责接收Client提交的做业，管理Worker，并命令Worker启动Driver和Executor。 
Worker：Standalone模式中slave节点上 的 守护进程 ，负责管理本节点的资源，按期向 Master汇报心跳，接收Master的命令，启动Driver和Executor 
Driver： 一个Spark做业运行时包括一个Driver进程，也是做业的主进程，负责做业的解析、生成Stage并调度Task到Executor上。包括DAGScheduler，TaskScheduler。 
Executor：即真正执行做业的地方，一个集群通常包含多个Executor，每一个Executor接收Driver的命令Launch Task，一个Executor能够执行一到多个Task。 
Stage：一个Spark做业通常包含一到多个Stage。 
Task：一个Stage包含一到多个Task，经过多个Task实现并行运行的功能 
DAGScheduler： 实现将Spark做业分解成一到多个Stage，每一个Stage根据RDD的Partition个数决定Task的个数，而后生成相应的Task set放到TaskScheduler中。 
TaskScheduler：实现Task分配到Executor上执行。 
提交做业有两种方式，分别是Driver（做业的master，负责做业的解析、生成stage并调度task到，包含DAGScheduler）运行在Worker上，Driver运行在客户端。接下来分别介绍两种方式的做业运行原理。

Driver运行在Worker上

经过org.apache.spark.deploy.Client类执行做业，做业运行命令以下：

./bin/spark-class org.apache.spark.deploy.Client launch spark://host:port file:///jar_url org.apache.spark.examples.SparkPi spark://host:port

• 1

做业执行流程描述：

一、客户端提交做业给Master 
二、Master让一个Worker启动Driver，即SchedulerBackend。Worker建立一个DriverRunner线程，DriverRunner启动SchedulerBackend进程。 
三、另外Master还会让其他Worker启动Exeuctor，即ExecutorBackend。Worker建立一个ExecutorRunner线程，ExecutorRunner会启动ExecutorBackend进程。 
四、ExecutorBackend启动后会向Driver的SchedulerBackend注册。SchedulerBackend进程中包含DAGScheduler，它会根据用户程序，生成执行计划，并调度执行。对于每一个stage的task，都会被存放到TaskScheduler中，ExecutorBackend向SchedulerBackend汇报的时候把TaskScheduler中的task调度到ExecutorBackend执行。 
五、全部stage都完成后做业结束。

Driver运行在客户端

./bin/run-example org.apache.spark.examples.SparkPi spark://host:port

• 1

 
一、客户端启动后直接运行用户程序，启动Driver相关的工做：DAGScheduler和BlockManagerMaster等。 
二、客户端的Driver向Master注册。 
三、Master还会让Worker启动Exeuctor。Worker建立一个ExecutorRunner线程，ExecutorRunner会启动ExecutorBackend进程。 
四、ExecutorBackend启动后会向Driver的SchedulerBackend注册。Driver的DAGScheduler解析做业并生成相应的Stage，每一个Stage包含的Task经过TaskScheduler分配给Executor执行。 
五、全部stage都完成后做业结束。

基于Yarn的Spark架构与做业执行流程

 
这里Spark AppMaster至关于Standalone模式下的SchedulerBackend，Executor至关于standalone的ExecutorBackend，spark AppMaster中包括DAGScheduler和YarnClusterScheduler。

refrences：

http://www.tuicool.com/articles/qaEVFb 
http://www.cnblogs.com/gaoxing/p/5041806.html 
http://blog.csdn.net/stark_summer/article/details/42833609