Spark的管理精髓

Posted 2021-04-03 大数据挖呀挖

Spark中有几套不同的架构，学习起来容易混淆。

我在学习的过程中，发现类比到公司的管理上，都可以对号入座。

想想也是，spark和公司，都是分布式的管理，都要考虑：

怎么调度资源？

怎么管理资源？

怎么把项目和组织分开，形成矩阵式管理？

这么一看，spark的所有概念就都很好理解了。

 

接下来就对spark的四对概念进行解读一下：

1. 客户端 / 集群：

2. driver / executor

3. master / worker

4. worker / executor

客户端 / 集群

客户端的主要作用是负责启动和管理driver，在公司里类似于项目管理团队。Spark集群主要是负责调度资源进行干活的，更多像技术团队。客户端通过启动application，向master提交任务。

master / worker

master是技术团队的大boss，需要了解各个团队(worker)的资源分布情况，以便于分配任务。还要定时检测有没有哪个worker不工作了，如果检测到哪个worker不工作了，就不再给其分配任务。所以master里会有很多定时任务，大致类比到公司里的例会，大boss需要通过经常的开会来了解下面团队的活跃情况。Worker可以类比到团队leader，worker的手下还有真正负责搬砖的executor。

worker / executor

worker是有资源的，这个资源就是内存(memory)和CPU cores。当master发指令给worker的时候，worker会启executor。

 

关键点：worker启动多少个小弟executor干活呢？

我们想一想，在公司里，一个新项目，项目经理找到了研发老大，研发老大发指令给几个部门leader，每个部门leader安排几个小弟干活是取决于什么因素呢？

答案可不是worker自己拍脑袋决定的，而是取决于项目的要什么样的资源。

比如项目经理说了，我这个项目，每个executor必须要有1g的内存，低于1g的executor是干不了我这个活的。那么如果每个worker有4g的内存，就只能启动4个executor，也就是派出4个小弟来参与到项目。

所以，每个worker启动几个executor的因素有：

1. 集群的总内存

2. 每个executor的内存

例如，我之前试验的环境是2台机器共32g的内存，一开始指定了spark.executor.memory=8g，每次运行只能起4个executor。后来修改了指定每个executor的内存数量是1g，每次就可以看到32个executors在干活。

 

说到内存，就还得提一下它的好兄弟：cores。

上面说到通过指定executor的内存，决定了会启动多少个executor。

不妨假设每个spark.executor.memory=1g，启动了32个executor。

如果集群共16个cores，而客户端又指定了一个executor需要1个core。那显然是不够分的。所以executor指定的core并不是物理的core，而是逻辑核。这种情况下，一个CPU的物理核对应2个executor的逻辑核。

所以，每个executor的逻辑核于CPU的物理核的对应关系，是由以下因素决定的：

1. Executor的数量。

2. spark.cores.max，如果没有设置，默认是使用集群的所有资源，即物理GPU核数。

3. spark.executor.cores。

但是，通常的规则是cpu的逻辑核数要等于cpu的物理核数。

这样一来，整个的设计顺序应该如下：

1. 设置spark.executor.cores=1，也就是每一个executor分配一个物理CPU的核。

2. 根据物理核数，决定executor的数量。例如CPU有16个核，那么共有16个executor。

3. 根据总的内存和executor的数量，设置：spark.executor.memory = 集群总memory / 物理CPU核数。

 

 

 

为啥要讨论executor的逻辑核呢？因为这要牵扯出spark里最重要的一个概念：task的并行度。

 

Task相关概念：

• Job：Job是由Action触发的，因此一个Job包含一个Action和N个Transform操作。例如df.show()就是一个标准的action操作，pyspark是惰性的，只有遇到action才开始触发所有的执行，所以调试程序时，会发现前边的transform操作都很快，遇到一个show()或者collect()会执行很长时间。

• Stage：Stage是由于shuffle操作而进行划分的Task集合，Stage的划分是根据其宽窄依赖关系；窄依赖的意思是每个分区的操作不涉及到其他分区；宽依赖的意思是当前分区的操作还涉及到其他分区。例如map是个窄依赖，join是个宽依赖。

• Task：最小执行单元，因为每个Task只是负责一个分区的数据处理，因此一般有多少个分区就有多少个Task，这一类的Task其实是在不同的分区上执行一样的动作。

如下图是一个job，被划分成了3个stage，每个stage的结束都会出现宽依赖。 

 

 

资源和任务是分布式计算的正反面。分布式最大的优势是任务的并行，并行度取决于任务的partition和executor的核数。

Partition是指RDD的分区数量，也就是把输入的文件分成了多少份。

一个partition对应一个task，而一个task只能分配一个executor的一个逻辑核。

Case1: 100个partition，只启动了10个executor，一个executor只有2个逻辑核。那么每一轮只能同时跑20个task，100个partition需要5轮才能完成。

Case2: 10个partition，有10个executor，一个executor有2个逻辑核。那么每一轮只有10个executor的cores在干活，其他一半资源轮空，无活可干。

 

所以，设计一个并行度比较好的任务的决定因素有：

1. 物理资源。根据上面的讨论，通用的规则是使得executor的总核数=CPU的物理总核数，所以资源方面的并行度是由CPU的物理总核数决定的。

2. 任务并行。任务是由RDD的partition数量决定的。指导原则是使得RDD的partition数量是CPU的物理总核数的整数倍，这样每一轮都能发挥所有的CPU火力。Pyspark提供了repartition函数可以重新修改分区的数量。

 

 driver / executor

在领导们都沟通完毕后，真正干活的两个角色登场了，他们就是负责解析任务、划分任务的项目经理driver，和负责搬砖的executor。

driver负责将你写的程序进行解析，构建有向无环图DAG，划分为一个个job、stage、task。如下图所示是一个构建完的DAG。

 

然后executor带着分配好的memory和cpu cores扑向一个个task，如果一轮搞不完，就再执行一轮。

写在最后

本文主要是以四对概念为脉络，将spark中最核心的架构进行了梳理，希望能对大家入门spark有所帮助。