spark学习之资源调度

&＃x1f48c;前一篇博客中&＃xff0c;我们学习了spark代码的执行过程&＃xff0c;其中涉及到了逻辑执行计划和物理执行计划&＃xff0c;今天我们主要来学习spark的资源调度的内容&＃xff0c;对往期内容感兴趣的同学可以参考如下内容&＃x1f447;:

• 上一篇: spark学习之执行计划explain.
• hadoop专题: hadoop系列文章.
• spark专题: spark系列文章.
• flink专题: Flink系列文章.

&＃x1f421;说到spark的资源调度&＃xff0c;我们主要关心的是执行计算任务节点的资源调度的设置&＃xff0c;如单个Executor的核数、单个节点Executor的个数、单个Executor的内存大小等&＃xff0c;接下来将会对这些进行详细的介绍。

1. 总体资源

我们这里假设单台服务器的内存大小为128g,32个线程。

1.1 Executor的核数

executor-cores的个数决定任务的并行度&＃xff0c;也就是同时执行task的个数&＃xff0c;一般情况下&＃xff0c;executor-cores的个数设置为3&＃xff5e;6个之间比较合适。

1.2 Executor的个数

这里的Executor的个数是指整个集群的Executor的个数个数&＃xff0c;所以&＃xff1a;
$$总的executor个数\&＃61;每个节点的executor数\ast work节点数$$
那么单个节点的Executor个数如何计算&＃xff0c;单个节点所能接受的最大Executor个数和yarn给的资源和Executor的核数有关&＃xff0c;关系如下&＃xff1a;

每个node的executor数&＃61;单节点yarn总核数每个executor的最大cpu核数每个 node 的 executor 数 &＃61; \dfrac{单节点 yarn 总核数}{每个 executor 的最大 cpu 核数} 每个node的executor数&＃61;每个executor的最大cpu核数单节点yarn总核数​

考虑到系统基础服务和 HDFS 等组件的余量&＃xff0c;yarn的nodemanager资源配置为&＃xff1a;28&＃xff0c;参数 executor-cores 的值为&＃xff1a;4&＃xff0c;那么每个 node 的 executor 数 &＃61; 28/4 &＃61; 7,假设集群节点为 10&＃xff0c;那么 num-executors &＃61; 7 * 10 &＃61; 70

1.3 Executor的内存

每个Executor的内存的大小也和yarn分配的资源有关系&＃xff1a;
executor内存大小&＃61;yarn总内存大小每个节点的executor数量executor内存大小&＃61; \dfrac{yarn总内存大小}{每个节点的 executor 数量} executor内存大小&＃61;每个节点的executor数量yarn总内存大小​
例如&＃xff1a;单个节点的 yarn 的参数配置为 100G&＃xff0c;那么每个 Executor 大概就是 100G/7≈14G,同时要注意yarn 配置中每个容器允许的最大内存是否匹配&＃xff0c;一般情况yarn默认配置的每个容器的内存大小范围为[1g,8g].

2. 内存资源

我们先来看一下spark的内存划分&＃xff1a;

• storage空间&＃xff1a;用来存放cache、persist和广播变量的缓存数据 。&＃xff08;Storage 内存 &＃61; 广播变量 &＃43; cache/Executor 数量&＃xff09;
• executor空间&＃xff1a;主要负责执行计算过程中的内存开销&＃xff0c;比如shuffle过程需要的内存。&＃xff08;Executor 内存 &＃61; 每个 Executor 核数 * &＃xff08;数据集大小/并行度&＃xff09;&＃xff09;
• other空间&＃xff1a;自定义的数据结构和元数据&＃xff0c;&＃xff08;Other 内存 &＃61; 自定义数据结构*每个 Executor 核数&＃xff09;

3. 持久化和序列化

我们先来介绍一下持久化和序列化在spark中的含义&＃xff1a;

• 持久化&＃xff1a;在不同操作间&＃xff0c;持久化&＃xff08;或缓存&＃xff09;一个数据集在内存中。当你持久化一个RDD&＃xff0c;每一个结点都将把它的计算分块结果保存在内存中&＃xff0c;并在对此数据集&＃xff08;或者衍生出的数据集&＃xff09;进行的其它动作中重用。这将使得后续的动作&＃xff08;action&＃xff09;变得更加迅速&＃xff08;通常快10倍&＃xff09;。
• 序列化&＃xff1a;序列化是将对象的状态信息转换为可以存储或传输的形式的过程&＃xff0c;主要目的有两个&＃xff1a;1. 存储到磁盘&＃xff1b; 2. 通过网络进行传输&＃xff1b;而在spark中&＃xff0c;算子相关的操作在Excutor上执行&＃xff0c;算子之外的代码在Driver端执行&＃xff0c;在执行有些算子的时候&＃xff0c;需要只用到Driver里面定义的数据&＃xff0c;这就涉及到了跨进程或着跨节点之间的通讯&＃xff0c;所以要求传递给Excutor中的数组所属的类型必须实现Serializable接口。

3.1 RDD

我们使用rdd持久化时&＃xff0c;默认 cache 缓存级别&＃xff08;memory_only&＃xff09;&＃xff0c;如果此时存储内存较小&＃xff0c;可以采用kryo&＃43;序列化缓存&＃xff0c;可以优化存储内存占用。

3.2 Dataset和Dataframe

我们使用dataset持久化时&＃xff0c;默认cache缓存级别&＃xff08;memory_and_disk&＃xff09;,即内存不够磁盘来凑&＃xff0c;如果采用序列化的方式&＃xff0c;实际效果和未序列化差别不大&＃xff0c;因为Dataset和Dataframe是被优化过的rdd&＃xff0c;所以开发中直接使用cache缓存即可。

总体来说&＃xff0c;如果使用RDD进行持久化&＃xff0c;建议采用kryo序列化&＃43;持久化的操作&＃xff0c;如果使用Dataset和Dataframe直接使用cache持久化即可。从性能上来讲&＃xff0c;DataSet,DataFrame 优于 RDD&＃xff0c;建议开发中使用 DataSet、DataFrame。

4. CPU资源

首先来了解几个非常容易搞混的概念&＃xff1a;

4.1 并行度

并行度指的就是task的数量&＃xff0c;或者说分区数量

• rdd的控制方法&＃xff1a;spark.default.parallelism&＃xff0c;根据算子计算决定
• sql的控制方法&＃xff1a;spark.sql.shuffle.partitions&＃xff0c;默认200

4.2 并发度

并发度指的可同时执行的最大task数量&＃xff0c;那这和并行度有何区别&＃xff0c;首先并行度是指一个任务需要多少个分区&＃xff0c;比如groupby之后产生了200个分区&＃xff08;200task&＃xff09;&＃xff0c;但是我们的executor只有2个核&＃xff0c;即我们最多同时执行2个task&＃xff0c;那么这个任务的并发度就是2&＃xff0c;并行度是200。每次执行2个task&＃xff0c;执行完后换上下一个task&＃xff0c;一直到执行完200个。

4.3 合理利用CPU

下面两种情况会导致CPU效率过低&＃xff1a;

• 并行度较低、数据分片较大容易导致 CPU 线程挂起&＃xff08;处理数据量大&＃xff09;
• 并行度过高、数据过于分散会让调度开销更多&＃xff08;task数量过多&＃xff09;

根据经验&＃xff1a;一般会将并行度&＃xff08;task 数&＃xff09;设置成并发度 &＃xff08;vcore 数&＃xff09;的 2 倍到 3 倍。

这一部分的详细内容可参考: spark学习之并行度、并发、core数和分区的关系.

5. 总结

本博客主要介绍了spark运行过程中的一些资源调度的情况&＃xff0c;以及如何配置资源调度使spark的运行效率得到提高&＃xff0c;主要从内存、持久化、cpu资源等方面进行了介绍。

6. 参考文章

• 《尚硅谷大数据技术之 Spark 调优》
• 《spark权威指南》
• 链接: spark持久化.
• 链接: 序列化的定义.