Spark原理详解

r囧r小猫 2022-05-30 03:55 291阅读 0赞

Hadoop存在缺陷：

基于磁盘，无论是MapReduce还是YARN都是将数据从磁盘中加载出来，经过DAG，然后重新写回到磁盘中

计算过程的中间数据又需要写入到HDFS的临时文件

这些都使得Hadoop在大数据运算上表现太“慢”，Spark应运而生。

Spark的架构设计：

![20180304184425303][]

ClusterManager负责分配资源，有点像YARN中ResourceManager那个角色，大管家握有所有的干活的资源，属于乙方的总包。

WorkerNode是可以干活的节点，听大管家ClusterManager差遣，是真正有资源干活的主。

Executor是在WorkerNode上起的一个进程，相当于一个包工头，负责准备Task环境和执行Task，负责内存和磁盘的使用。

Task是施工项目里的每一个具体的任务。

Driver是统管Task的产生与发送给Executor的，是甲方的司令员。

SparkContext是与ClusterManager打交道的，负责给钱申请资源的，是甲方的接口人。

整个互动流程是这样的：

1 甲方来了个项目，创建了SparkContext，SparkContext去找ClusterManager申请资源同时给出报价，需要多少CPU和内存等资源。ClusterManager去找WorkerNode并启动Excutor，并介绍Excutor给Driver认识。

2 Driver根据施工图拆分一批批的Task，将Task送给Executor去执行。

3 Executor接收到Task后准备Task运行时依赖并执行，并将执行结果返回给Driver

4 Driver会根据返回回来的Task状态不断的指挥下一步工作，直到所有Task执行结束。

再看下图加深下理解：

![20180304184644505][]

Spark的核心组件：

![20180304184711707][]

核心部分是RDD相关的，就是我们前面介绍的任务调度的架构，后面会做更加详细的说明。

SparkStreaming：

基于SparkCore实现的可扩展、高吞吐、高可靠性的实时数据流处理。支持从Kafka、Flume等数据源处理后存储到HDFS、DataBase、Dashboard中。

MLlib：

关于机器学习的实现库，关于机器学习还是希望花点时间去系统的学习下各种算法，这里有一套基于Python的ML相关的博客教材http://blog.csdn.net/yejingtao703/article/category/7365067。

SparkSQL：

Spark提供的sql形式的对接Hive、JDBC、HBase等各种数据渠道的API，用Java开发人员的思想来讲就是面向接口、解耦合，ORMapping、Spring Cloud Stream等都是类似的思想。

GraphX：

关于图和图并行计算的API，我还没有用到过。

RDD(Resilient Distributed Datasets) 弹性分布式数据集

RDD支持两种操作：转换（transiformation）和动作（action）

转换就是将现有的数据集创建出新的数据集，像Map；动作就是对数据集进行计算并将结果返回给Driver，像Reduce。

RDD中转换是惰性的，只有当动作出现时才会做真正运行。这样设计可以让Spark更见有效的运行，因为我们只需要把动作要的结果送给Driver就可以了而不是整个巨大的中间数据集。

缓存技术（不仅限内存，还可以是磁盘、分布式组件等）是Spark构建迭代式算法和快速交互式查询的关键，当持久化一个RDD后每个节点都会把计算分片结果保存在缓存中，并对此数据集进行的其它动作（action）中重用，这就会使后续的动作（action）变得跟迅速（经验值10倍）。例如RDD0àRDD1àRDD2，执行结束后RDD1和RDD2的结果已经在内存中了，此时如果又来RDD0àRDD1àRDD3，就可以只计算最后一步了。

RDD之间的宽依赖和窄依赖：

窄依赖：父RDD的每个Partition只被子RDD的一个Partition使用。

宽依赖：父RDD的每个Partition会被子RDD的多个Partition使用。

宽和窄可以理解为裤腰带，裤腰带扎的紧下半身管的严所以只有一个儿子；裤腰带帮的比较宽松下半身管的不禁会搞出一堆私生子，这样就记住了。

对于窄依赖的RDD，可以用一个计算单元来处理父子partition的，并且这些Partition相互独立可以并行执行；对于宽依赖完全相反。

在故障回复时窄依赖表现的效率更高，儿子坏了可以通过重算爹来得到儿子，反正就这一个儿子当爹的恢复效率就是100%。但是对于宽依赖效率就很低了，如下图：

![20180304184752723][]

如果儿子b1坏了a1、a2、a3三个当爹的都运算了一次恢复了b1，但是其实它们的运算同时也会覆盖一遍b2这个无辜的儿子，有效率只有50%。

代码实现上窄依赖NarrowDependency有2种：OneToOneDependency和RangeDependency

宽依赖只有1种ShuffleDependency，但是内部参数ShuffleManager有Hash和Sort两种，后面会详细介绍Hash和Sork的区别。

有了以上RDD宽窄依赖和父子之间的血缘关系，我们就可以绘制DAG：

绘制原则就是由于宽依赖的“断点”效应，根据宽依赖将整个DAG分为不同的阶段（Stage），每个Stage之间有先后关系从前向后进行，在每个Stage内部窄依赖RDD是并行执行的。

Stage的划分是从最后一个RDD从后往前进行的。

**注意：转化和动作只是决定惰性执行的时机，宽窄依赖才是划分Stage的唯一标准**。ReduceByKey是转化，但它包含ShuffleDependency，所以转化和动作与宽窄依赖没关，不要混淆。

RDD的计算：

Spark的Task有两种：ShuffleMapTask和ResultTask，其中后者在DAG最后一个阶段推送给Executor，其余所有阶段推送的都是ShuffleMapTask。

Executor在准备好Task运行环境后会调用scheduler.Task\#run，scheduler.Task\#run会调用ShuffleMapTask或ResultTask的runTask，runTask会调用RDD的\#iterator，每个RDD真正的计算逻辑实现在RDD的computer方法中。用户创建SparkContext时会创建SparkEnv负责管理所有运行环境的信息，最核心的是cacheManager。

CheckPoint：

CheckPoint是对RDD缓存不足被擦写等中间block断丢失导致重新计算这一缺点的弥补，CheckPoint会启动一个job来计算并将计算结果写入磁盘中，最后修改原始RDD的依赖为当前CheckPoint。当缓存没有命中时先来看CheckPoint中有没有记录，再决定是否重新计算。CheckPoint是RDD磁盘缓存的一种表现，稳定性更高，但是IO更慢。

Spark任务调度模块DAGScheduler、TaskScheduler：

![20180304184829371][]

用户编排的代码由一个个的RDD Objects组成，DAGScheduler负责根据RDD的宽依赖拆分DAG为一个个的Stage，买个Stage包含一组逻辑完全相同的可以并发执行的Task。TaskScheduler负责将Task推送给从ClusterManager那里获取到的Worker启动的Executor。

DAGScheduler（统一化的，Spark说了算）：

详细的案例分析下如何进行Stage划分，请看下图

![20180304184902093][]

1 stage是触发action的时候从后往前划分的，所以本图要从RDD\_G开始划分。

2 RDD\_G依赖于RDD\_B和RDD\_F，随机决定先判断哪一个依赖，但是对于结果无影响。

3 RDD\_B与RDD\_G属于窄依赖，所以他们属于同一个stage，RDD\_B与老爹RDD\_A之间是宽依赖的关系，所以他们不能划分在一起，所以RDD\_A自己是一个stage1

4 RDD\_F与RDD\_G是属于宽依赖，他们不能划分在一起，所以最后一个stage的范围也就限定了，RDD\_B和RDD\_G组成了Stage3

5 RDD\_F与两个爹RDD\_D、RDD\_E之间是窄依赖关系，RDD\_D与爹RDD\_C之间也是窄依赖关系，所以他们都属于同一个stage2

6 执行过程中stage1和stage2相互之间没有前后关系所以可以并行执行，相应的每个stage内部各个partition对应的task也并行执行

7 stage3依赖stage1和stage2执行结果的partition，只有等前两个stage执行结束后才可以启动stage3.

8 我们前面有介绍过Spark的Task有两种：ShuffleMapTask和ResultTask，其中后者在DAG最后一个阶段推送给Executor，其余所有阶段推送的都是ShuffleMapTask。在这个案例中stage1和stage2中产生的都是ShuffleMapTask，在stage3中产生的ResultTask。

9 虽然stage的划分是从后往前计算划分的，但是依赖逻辑判断等结束后真正创建stage是从前往后的。也就是说如果从stage的ID作为标识的话，先需要执行的stage的ID要小于后需要执行的ID。就本案例来说，stage1和stage2的ID要小于stage3，至于stage1和stage2的ID谁大谁小是随机的，是由前面第2步决定的。

虽然理论上Task应该交给workerNode上的executor来执行的，但是有一种情况下是是在DAG划分结束后直接在本地执行的。

1 Spark.localExecution.enabled设置为true；

2 用户显示指定允许本地执行；

3 整个DAG只有一个stage；

4 仅有一个Partition。

同时满足上面4个条件下，可以直接在SparkContext（Driver）节点上本地执行。

TaskScheduler（接口化的，根据不同的部署方式Standalone、Mesos、YARN、Local）：

每个TaskScheduler对应着一个帮手SchedulerBackend，SchedulerBackend负责与ClusterManager交互获得资源，然后将这些资源信息传给TaskScheduler，TaskScheduler负责监督Task的执行状态并进行相应的调度。这里主要做的工作有：就近原则、失败重试、慢任务推测性执行

任务调度的时候默认是FIFO（先到先得）的，由jobID和stageID的大小来决定；也可以配置成FAIR（公平）的，重新确定调度顺序推送task给Executor。

Executor执行完Task后会通过向Driver发送StatusUpdate的消息来通知Driver任务更新Task的状态。Driver会将Task状态通知转告给TaskSchedule，后者会重新分配计算任务。

假如Task有执行失败的，根据失败原因和阈值进行该Task的重试或者放弃。

假如所有Task执行成功，如果Task是ResultTask，那么任务结束；如果是ShuffleMapTask那么启动下一个stage。

Spark运行模式：

Local模式：

比较简单，只适用于自己玩和测试，甲方SparkContext乙方Executor等都部署在一起，物理位置上角色定位不明确。

Mesos模式：

![20180304184935249][]

Worker部分采用Master/Slaver模式，Master是整个系统的核心部件所以用ZooKeeper做高可用性加固，Slaver真正创建Executor执行Task并将自己的物理计算资源汇报给Master，Master负责将slavers的资源按照策略分配给Framework。

Mesos资源调度分为粗粒度和细粒度两种方式：

**粗粒度**方式是启动时直接向Master申请执行全部Task的资源，并等所有计算任务结束后才释放资源；**细粒度**方式是根据Task需要的资源不停的申请和归还。两个方式各有利弊，粗粒度的优点是调度成本小，但是会因木桶效应造成资源长期被霸占；细粒度没有木桶效应，但是调度上的管理成本较高。

YARN模式：

![20180304185004716][]

YRAN模式下分为Cluster和Client两种模式，上图中的为Cluster模式。

Cluster模式下就是将Spark作为一个普通的YARN任务，Client端通过ResourceManager申请到资源，创建ApplicationMaster、Task到Container中去。ApplicationMaster负责监督Task的执行情况。

Client模式与Cluster模式的不同点是Client模式下SparkContext是在client本机下创建运行的，client只通过ResourceManager申请资源在工作节点中执行Task，负责管理和监督的领导层没有交给YARN，还在Client本机。

Standalone（也叫Deploy）模式：

与Mesos模式有点像，也是Master/Slavers的架构。

Master是负责整个集群资源调度和Application管理的核心，需要高可用；

Slaver是干活的苦力，将自己注册给Master后接收Master资源分配和调度命令，启动Executor、执行Task；

Client负责Application的创建并向Master注册Application，接收来自Slaver的Task结果更新。

Spark的容错处理：

请注意这里使用的是容错而不是容灾，因为这俩不是一个概念。

**容灾**是洪水、火灾、地震等导致的灾难性的毁灭性的故障，是非常小概率的事件，需要做数据级别甚至应用级别的异地备份；而**容错**是解决由于网络阻塞、磁盘损坏、内存溢出、机器掉电等引起的单点故障或者模块化的故障，是每时每刻都有可能发生的大概率事件。

所以说容错和容灾不是一个级别的，我们对架构拓扑稍作优化通过很小的成本就可以达到容错的效果；但是要想达到容灾那将是巨大的开销而且很难存在一个100%容灾的设计，例如地球被炸了难道还要将数据在外太空做定期备份么~~

前面介绍模式的时候我们一再强调，Master是核心部件，是心脏和大脑，所以Master的故障是我们不能接受的，所以需要通过Zookeeper来做高可用性（虽然也有fileSystem模式但是自从我做过的一个产品真的出现磁盘损坏导致单点故障然后连续加班了40多小时候以后我再也不相信硬件了）。一旦Leader的Master挂掉，其它Master会自主推优出新的Leader。新的Leader会从ZooKeeper中读取所有元数据并通知到大家（Worker、Client）自己登基上位。

Worker节点众多，出现故障的概率最高，workers定时的向Master上报心跳，一旦超时Master将对它进行卸磨杀驴。Master会将worker上所有Executor设置为失效并通知给Client，Client会通知SchedulerBackend如果有该worker上的executor正在执行你的task请重新调度。Master通知完以后会尝试kill掉该坏掉的worker。

Worker节点上运行着很多executor进程，如果worker心跳没问题但是某个executor进程出线了问题怎么办？这个概率比worker出现异常的概率更大！其实worker节点除了明着干活的executor，还有监督executor的executorRunner，它会将executor退出的信息告知自己所在的worker，worker在通知自己的master，剩下的就是跟上面一样的套路了。

Executor详解：

Executor干了两件事情：运行Task和将结果反馈给Driver。。

Master在给Application分配Worker时有两种方式：尽量打散和尽量集中。

尽量打散适用于内存密集型，尽量集中适用于CPU密集型。

1个物理节点可以部署多个Worker，但是一个Worker中对于1个Application只能有1个Executor。

关于Executor的内存设置：

Executor是执行Task的真正苦力，内存设置的过小，会导致内存溢出或者频繁GC影响效率；内存设置过大会导致占用过多资源（内存资源从价格上和槽道数量上来讲还是比较珍贵的）。所以合理的设置Executor内存是Spark处理任务的关键。Executor支持的任务的数量取决于持有的CPU的核数，所以一种思路是如果集群普遍的CPU核数够多但是内存紧张，可以采用更多的分区来增加Task的个数减少单个Task执行对内存的要求。

PS：2017年内存条一年涨三倍，显卡也涨了两倍，这真是互联网+人工智能让整个产业一年暴富啊！

Executor最终将Task的执行结果反馈给Driver，会根据大小采用不同的策略：

1 如果大于MaxResultSize，默认1G，直接丢弃；

2 如果“较大”，大于配置的frameSize（默认10M），以taksId为key存入BlockManager

3 else，全部吐给Driver。

Shuffle详解：

Hash Base Shuffle（spark1.2以前默认）：

下图是将4个Partition洗牌成3个Partition的案例，假设当前是StageA，下一个是StageB

![20180304185039972][]

在洗牌过程中StageA每个当前的Task会把自己的Partition按照stageB中Partition的要求做Hash产生stageB中task数量的Partition（这里特别强调是每个stageA的task），这样就会有len（stageA.task）\*len(stageB.task)这么多的小file在中间过程产生，如果要缓存RDD结果还需要维护到内存，下个stageB需要merge这些file又涉及到网络的开销和离散文件的读取，所以说超过一定规模的任务用Hash Base模式是非常吃硬件的。

尽管后来Spark版本推出了Consolidate对基于Hash的模式做了优化，但是只能在一定程度上减少block file的数量，没有根本解决上面的缺陷。

Sort Base Shuffle（spark1.2开始默认）：

![20180304185112227][]

Sort模式下StageA每个Task会产生2个文件：内容文件和索引文件。内容文件是根据StageB中Partition的要求自己先sort好并生成一个大文件；索引文件是对内容文件的辅助说明，里面维护了不同的子partition之间的分界，配合StageB的Task来提取信息。这样中间过程产生文件的数量由len（stageA.task）\*len(stageB.task)减少到2\* len(stageA.task)，StageB对内容文件的读取也是顺序的。Sort带来的另一个好处是，一个大文件对比与分散的小文件更方便压缩和解压，通过压缩可以减少网络IO的消耗。（PS：但是压缩和解压的过程吃CPU，所以要合理评估）

Sort和Hash模式通过spark.shuffle.manager来配置的。

Storage模块：

存储介质：内存、磁盘、Tachyon（这货是个分布式内存文件，与Redis不一样，Redis是分布式内存数据库），存储级别就是它们单独或者相互组合，再配合一些容错、序列化等策略。例如内存+磁盘。

负责存储的组件是BlockManager，在Master（Dirver）端和Slaver（Executor）端都有BlockManager，分工不同。Slaver端的将自己的BlockManager注册给Master，负责真正block；Master端的只负责管理和调度。

Storage模块运行时内存默认占Executor分配内存的60%，所以合理的分配Executor内存和选择合适的存储级别需要平衡下Spark的性能和稳定。

结束语：Spark的出现很好的弥补了Hadoop在大数据处理上的不足，同时随着枝叶不断散开，出线了很多的衍生的接口模块丰富了Spark的应用场景，也降低了Spark与其他技术的接入门槛。

[20180304184425303]: /images/20220530/c953115b17d84b938cbac8aa66529996.png
[20180304184644505]: /images/20220530/9a1ec7ba158f495cbb8fabe5aa82f81c.png
[20180304184711707]: /images/20220530/c6c49c1bd31a4aaf869cbd47474c176a.png
[20180304184752723]: /images/20220530/c3f33c418aac443894d4781fbb5f20fa.png
[20180304184829371]: /images/20220530/37ea3c7a36734899be181545afa29a58.png
[20180304184902093]: /images/20220530/879b9ffea36e43828eb9fbf58c348803.png
[20180304184935249]: /images/20220530/d3c56abace5047ae927f1038948700fb.png
[20180304185004716]: /images/20220530/73f43838ffca484d878ad142d2d0d1e4.png
[20180304185039972]: /images/20220530/68c3816e840b48c7b88d654146bc36b6.png
[20180304185112227]: /images/20220530/62db02ea4d6c4ddc8f085f63cbbd4c9e.png