Spark性能调优实战(基础知识)-极客时间-吴磊

我不是女神ヾ 2022-10-16 15:21 390阅读 0赞

课程连接：  
https://time.geekbang.org/column/intro/400

# 一.调优的方法论 #

# 1.spark调优整思想 #

## 性能调优的目的 ##

性能调优的最终目的，是在所有参与计算的硬件资源之间寻求协同与平衡，让硬件资源达到一种平衡、无瓶颈的状态。不能一直这么无限循环下去。  
执行性能最好（运行时间最短）任务并不是那些把 CPU 利用率压榨到 100%，以及把内存设置到最大的配置组合，而是那些硬件资源配置最均衡的计算任务。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]

## 调优的主要套路 ##

以性能为导向的开发习惯，开发者可以按图索骥地去开展性能调优工作，做到有的放矢、事半功倍  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]  
性能篇我主要分两部分来讲

1.  \*\*通用调优篇：\*\*通用技巧，包括应用开发的基本原则、配置项的设置、Shuffle 的优化，以及资源利用率的提升。
2.  \*\*sparkSql调优篇：\*\*另一部分我会专注于数据分析领域，借助如 Tungsten、AQE 这样的 Spark 内置优化项和数据关联这样的典型场景，来和你聊聊 Spark SQL 中的调优方法和技巧。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]

# 2.性能调优的本质，该从哪里入手 #

## UDF 与 SQL functions 的区别 ##

Spark SQL 的 Catalyst Optimizer 能够明确感知 SQL functions 每一步在做什么，因此有足够的优化空间。相反，UDF（User Defined Functions） 里面封装的计算逻辑对于 Catalyst Optimizer 来说就是个黑盒子，除了把 UDF 塞到闭包里面去，也没什么其他工作可做的。

从ETL 应用中随意挑出一个自定义的 UDF，尝试用 SQL functions 去重写，然后准备单元测试数据，最后在单机环境下对比两种不同实现的运行时间。通常情况下，UDF 实现相比 SQL functions 会慢 3% 到 5% 不等。所以，UDF 的性能开销还是有的。

## 最短木桶思想 ##

使用 SQL functions 优化 UDF 的时候，为什么没有显著提升端到端 ETL 应用的整体性能呢？根据木桶理论，最短的木板决定了木桶的容量，因此，对于一只有短板的木桶，其他木板调节得再高也无济于事，最短的木板才是木桶容量的瓶颈。对于 ETL 应用这支木桶来说，UDF 到 SQL functions 的调优之所以对执行性能的影响微乎其微，根本原因在于它不是那块最短的木板。

## 性能调优的本质我们可以归纳为 4 点 ##

1.性能调优不是一锤子买卖，补齐一个短板，其他板子可能会成为新的短板。因此，它是一个动态、持续不断的过程。  
2. 性能调优的手段和方法是否高效，取决于它针对的是木桶的长板还是瓶颈。针对瓶颈，事半功倍；针对长板，事倍功半。  
3. 性能调优的方法和技巧，没有一定之规，也不是一成不变，随着木桶短板的此消彼长需要相应的动态切换。  
4. 性能调优的过程收敛于一种所有木板齐平、没有瓶颈的状态。

## 定位性能瓶颈的途径有哪些？ ##

一是先验的专家经验，二是后验的运行时诊断。

### 专家经验 ###

1.  所谓专家经验是指在代码开发阶段、或是在代码 Review 阶段，凭借过往的实战经验就能够大致判断哪里可能是性能瓶颈。

### 运行时诊断 ###

1.  运行时诊断的手段和方法应该说应有尽有、不一而足。比如：对于任务的执行情况，Spark UI 提供了丰富的可视化面板，来展示 DAG、Stages 划分、执行计划、Executor 负载均衡情况、GC 时间、内存缓存消耗等等详尽的运行时状态数据；对于硬件资源消耗，开发者可以利用 Ganglia 或者系统级监控工具，如 top、vmstat、iostat、iftop 等等来实时监测硬件的资源利用率；特别地，针对 GC 开销，开发者可以将 GC log 导入到 JVM 可视化工具，从而一览任务执行过程中 GC 的频率和幅度。

**定位瓶颈：从硬件资源消耗的角度切入，往往是个不错的选择：**  
从硬件的角度出发，计算负载划分为计算密集型、内存密集型和 IO 密集型。如果我们能够明确手中的应用属于哪种类型，自然能够缩小搜索范围，从而更容易锁定性能瓶颈。

## 性能调优的方法与手段 ##

定位到了性能瓶颈所在，那么，具体该如何调优呢？  
Spark 的性能调优可以从应用代码和 Spark 配置项这 2 个层面展开。

### 应用代码 ###

### Spark 配置项 ###

# 二.调优必备基础知识 #

# 3.RDD(弹性分布式数据集) #

Spark 中的核心概念 RDD 和 DAG  
RDD 可以说是 Spark 中最基础的概念了，使用 Spark 的开发者想必对 RDD 都不陌生，甚至提起 RDD，你的耳朵可能都已经听出茧子了。不过，随着 Spark 开发 API 的演进和发展，现在上手开发基本都是 DataFrame 或 Dataset API。所以很多初学者会认为，“反正 RDD API 基本都没人用了，我也没必要弄明白 RDD 到底是什么。”

尽管 RDD API 使用频率越来越低，绝大多数人也都已经习惯于 DataFrame 和 Dataset API，但是，无论采用哪种 API 或是哪种开发语言，你的应用在 Spark 内部最终都会转化为 RDD 之上的分布式计算。换句话说，如果你想要在运行时判断应用的性能瓶颈，前提是你要对 RDD 足够了解。

### RDD 的核心特征和属性 ###

RDD 具有 4 大属性，分别是 **partitions、partitioner(分区规则)、dependencies(血缘依赖关系) 和 compute(计算逻辑)** 属性。正因为有了这 4 大属性的存在，让 RDD 具有分布式(partitions+partitioner)和容错性(dependencies +compute)这两大最突出的特性。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]

#### partitions、partitioner 属性 ####

**在分布式运行环境中，RDD 封装的数据在物理上散落在不同计算节点的内存或是磁盘中，这些散落的数据被称“数据分片”，RDD 的分区规则决定了哪些数据分片应该散落到哪些节点中去。RDD 的 partitions 属性对应着 RDD 分布式数据实体中所有的数据分片，而 partitioner 属性则定义了划分数据分片的分区规则，如按哈希取模或是按区间划分等。**

**开发过程中应避免单机思维模式写代码，sprak中RDD是分布在集群中的个个节点中。**  
关于RDD的单机思维模式的例子：  
RDD中的集合排序问题  
RDD.collect.sort

#### dependencies 和 compute 属性 ####

在 Spark 中，任何一个 RDD 都不是凭空产生的，每个 RDD 都是基于某种计算逻辑从某个“数据源”转换而来。RDD 的 dependencies 属性记录了生成 RDD 所需的“数据源”，术语叫做父依赖（或父 RDD），compute 方法则封装了从父 RDD 到当前 RDD 转换的计算逻辑。  
容错：  
基于数据源和转换逻辑，无论 RDD 有什么差池（如节点宕机造成部分数据分片丢失），在 dependencies 属性记录的父 RDD 之上，都可以通过执行 compute 封装的计算逻辑再次得到当前的 RDD，如下图所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 4]

### preferredLocations ###

prefered location,选择在哪个RDD上取数据来计算最合适。

可以理解为最佳位置，移动计算不移动数据，数据在哪就在哪计算，减少IO，  
有些情况，可以数据不动代码动，有些情况，就必须要动数据了。  
preferredLocations不起作用的场景有一种是存算分离的场景，计算节点在k8s集群里，远程读取分布式文件系统，例如hdfs或者ceph等，可能接入万兆SDN网络能够有一些收益。

# 4.DAG与流水线：什么叫内存计算 #

## Spark 的内存计算的含义 ##

**在 Spark 中，内存计算有两层含义：第一层含义就是众所周知的分布式数据缓存，第二层含义是 Stage 内的流水线式计算模式。**

### 第一层含义：分布式数据缓存 ###

分布式数据缓存：Spark 允许开发者将分布式数据集缓存到计算节点的内存中，从而对其进行高效的数据访问。

RDD cache ：  
RDD cache 确实是 Spark 分布式计算引擎的一大亮点，但不能滥用 cache 机制，只有需要频繁访问的数据集才有必要 cache，对于一次性访问的数据集，cache 不但不能提升执行效率，反而会产生额外的性能开销，让结果适得其反。

### 第二层含义：Stage 内部的流水线式计算模式 ###

**Stage 内部的流水线式计算模式，不是简单地把数据和计算挪到内存，在 Spark 中，流水线计算模式指的是：在同一 Stage 内部，所有算子融合为一个函数，Stage 的输出结果由这个函数一次性作用在输入数据集而产生。这也正是内存计算的第二层含义。**

\*\*DAG 全称 Direct Acyclic Graph，中文叫有向无环图。\*\*顾名思义，DAG 是一种“图”。我们知道，任何一种图都包含两种基本元素：顶点（Vertex）和边（Edge），顶点通常用于表示实体，而边则代表实体间的关系。**在 Spark 的 DAG 中，顶点是一个个 RDD，边则是 RDD 之间通过 dependencies 属性构成的父子关系。**

从开发者的视角出发，DAG 的构建是通过在分布式数据集上不停地调用算子来完成的。

**Stages 的划分:**  
DAG 毕竟只是一张流程图，Spark 需要把这张流程图转化成分布式任务，才能充分利用分布式集群并行计算的优势。

从开发者构建 DAG，到 DAG 转化的分布式任务在分布式环境中执行，其间会经历如下 4 个阶段：

*  回溯 DAG 并划分 Stages
 *  在 Stages 中创建分布式任务
 *  分布式任务的分发
 *  分布式任务的执行  
    一个stage中的一个stak执行不结束这个stage就执行不结束，satage对应的job(任务)就执行不结束，及一颗老鼠屎坏了一锅粥。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 5]  
    **从 DAG 到 Stages 的转化过程是：以 Actions 算子为起点，从后向前回溯 DAG，以 Shuffle 操作为边界去划分 Stages。**  
    **Stage 中的内存计算:**

基于内存的计算模型并不是凭空产生的，而是根据前人的教训和后人的反思精心设计出来的。这个前人就是 Hadoop MapReduce，后人自然就是 Spark。

**MapReduce 提供两类计算抽象，分别是 Map 和 Reduce: 两个 Map 操作之间的计算，以及 Map 与 Reduce 操作之间的计算都是利用本地磁盘来交换数据的。不难想象，这种频繁的磁盘 I/O 必定会拖累用户应用端到端的执行性能。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 6]  
Spark的内存计算，不仅仅是指数据可以缓存在内存中，更重要的是通过计算的融合来大幅提升数据在内存中的转换效率，进而从整体上提升应用的执行性能。

**由于计算的融合只发生在 Stages 内部，而 Shuffle 是切割 Stages 的边界，因此一旦发生 Shuffle，内存计算的代码融合就会中断。但是，当我们对内存计算有了多方位理解以后，就不会一股脑地只想到用 cache 去提升应用的执行性能，而是会更主动地想办法尽量避免 Shuffle，让应用代码中尽可能多的部分融合为一个函数，从而提升计算效率。**

**shuffle的过程确实有落盘的步骤，但也仅限shuffle操作。stage内部是流水线式的内存计算，不会有落盘的动作。**

# 5.Spark调度系统：“数据不动代码动”含义 #

\*\*Spark 调度系统的原则是尽可能地让数据呆在原地、保持不动，同时尽可能地把承载计算任务的代码分发到离数据最近的地方，从而最大限度地降低分布式系统中的网络开销。\*\*毕竟，分发代码的开销要比分发数据的代价低太多，这也正是“数据不动代码动”这个说法的由来。

**Spark 调度系统的工作流程包含如下 5 个步骤：**

**1. 将 DAG 拆分为不同的运行阶段 Stages；  
2. 创建分布式任务 Tasks 和任务组 TaskSet；  
3. 获取集群内可用的硬件资源情况；  
4. 按照调度规则决定优先调度哪些任务 / 组；  
5. 依序将分布式任务分发到执行器 Executor。**

**Spark 调度系统包含 3 个核心组件**，分别是 DAGScheduler、TaskScheduler 和 SchedulerBackend。这 3 个组件都运行在 Driver 进程中，它们通力合作将用户构建的 DAG 转化为分布式任务，再把这些任务分发给集群中的 Executors 去执行。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 7]

### 1. DAGScheduler ###

DAGScheduler 的主要职责有二：

*  一是把用户 DAG 拆分为 Stages；
 *  二是在 Stage 内创建计算任务 Tasks，这些任务囊括了用户通过组合不同算子实现的数据转换逻辑。然后，执行器 Executors 接收到 Tasks，会将其中封装的计算函数应用于分布式数据分片，去执行分布式的计算过程。

不过，如果我们给集群中处于繁忙或者是饱和状态的 Executors 分发了任务，执行效果会大打折扣。因此，在分发任务之前，调度系统得先判断哪些节点的计算资源空闲，然后再把任务分发过去。那么，调度系统是怎么判断节点是否空闲的呢？

### 2. SchedulerBackend ###

SchedulerBackend 就是用来判断节点是否空闲，它是对于资源调度器的封装与抽象，为了支持多样的资源调度模式如 Standalone、YARN 和 Mesos，SchedulerBackend 提供了对应的实现类。在运行时，Spark 根据用户提供的 MasterURL，来决定实例化哪种实现类的对象。MasterURL 就是你通过各种方式指定的资源管理器，如 --master spark://ip:host（Standalone 模式）、–master yarn（YARN 模式）。

对于集群中可用的计算资源，SchedulerBackend 会用一个叫做 ExecutorDataMap 的数据结构，来记录每一个计算节点中 Executors 的资源状态。ExecutorDataMap 是一种 HashMap，它的 Key 是标记 Executor 的字符串，Value 是一种叫做 ExecutorData 的数据结构，ExecutorData 用于封装 Executor 的资源状态，如 RPC 地址、主机地址、可用 CPU 核数和满配 CPU 核数等等，它相当于是对 Executor 做的“资源画像”。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 8]  
总的来说，对内，SchedulerBackend 用 ExecutorData 对 Executor 进行资源画像；对外，SchedulerBackend 以 WorkerOffer 为粒度提供计算资源，WorkerOffer 封装了 Executor ID、主机地址和 CPU 核数，用来表示一份可用于调度任务的空闲资源。显然，基于 Executor 资源画像，SchedulerBackend 可以同时提供多个 WorkerOffer 用于分布式任务调度。WorkerOffer 这个名字起得蛮有意思，Offer 的字面意思是公司给你提供的工作机会，结合 Spark 调度系统的上下文，就变成了使用硬件资源的机会。

### 3. TaskScheduler ###

左边有需求，右边有供给，如果把 Spark 调度系统看作是一个交易市场的话，那么中间还需要有个中介来帮它们对接意愿、撮合交易，从而最大限度地提升资源配置的效率。在 Spark 调度系统中，这个中介就是 TaskScheduler。**TaskScheduler 的职责是，基于既定的规则与策略达成供需双方的匹配与撮合。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 9]  
TaskScheduler 的核心是任务调度的规则和策略，TaskScheduler 的调度策略分为两个层次，

*  一个是不同 **Stages 和Stages之间**的调度优先级，
 *  一个是 **Stages 内**不同任务之间的调度优先级。

**两个或多个 Stages：**  
两个或多个 Stages，如果它们彼此之间不存在依赖关系、互相独立，可以并行运行，在面对同一份可用计算资源的时候，它们之间就会存在竞争关系。

对于这种 Stages 之间的任务调度，TaskScheduler 提供了 2 种调度模式，分别是 FIFO（先到先得）和 FAIR（公平调度）

**Stages 内部：**  
Stages 内部的任务调度相对来说简单得多。当 TaskScheduler 接收到来自 SchedulerBackend 的 WorkerOffer 后，**TaskScheduler 会优先挑选那些满足本地性级别要求的任务进行分发**。众所周知，本地性级别有 4 种：Process local < Node local < Rack local < Any。从左到右分别是进程本地性、节点本地性、机架本地性和跨机架本地性。从左到右，计算任务访问所需数据的效率越来越差。

进程本地性表示计算任务所需的输入数据就在某一个 Executor 进程内，因此把这样的计算任务调度到目标进程内最划算。同理，如果数据源还未加载到 Executor 进程，而是存储在某一计算节点的磁盘中，那么把任务调度到目标节点上去，也是一个不错的选择。再次，如果我们无法确定输入源在哪台机器，但可以肯定它一定在某个机架上，本地性级别就会退化到 Rack local。

DAGScheduler 划分 Stages、创建分布式任务的过程中，会为每一个任务指定本地性级别，本地性级别中会记录该任务有意向的计算节点地址，甚至是 Executor 进程 ID。换句话说，任务自带调度意愿，它通过本地性级别告诉 TaskScheduler 自己更乐意被调度到哪里去。

既然计算任务的个人意愿这么强烈，TaskScheduler 作为中间商，肯定要优先满足人家的意愿。这就像一名码农想要租西二旗的房子，但是房产中介 App 推送的结果都是东三环国贸的房子，那么这个中介的匹配算法肯定有问题。

由此可见，Spark 调度系统的原则是尽可能地让数据呆在原地、保持不动，同时尽可能地把承载计算任务的代码分发到离数据最近的地方，从而最大限度地降低分布式系统中的网络开销。毕竟，分发代码的开销要比分发数据的代价低太多，这也正是“数据不动代码动”这个说法的由来。

**总的来说，TaskScheduler 根据本地性级别遴选出待计算任务之后，先对这些任务进行序列化。然后，交给 SchedulerBackend，SchedulerBackend 根据 ExecutorData 中记录的 RPC 地址和主机地址，再将序列化的任务通过网络分发到目的主机的 Executor 中去。最后，Executor 接收到任务之后，把任务交由内置的线程池，线程池中的多线程则并发地在不同数据分片之上执行任务中封装的数据处理函数，从而实现分布式计算。**

# 6.Spark 的存储系统：空间换时间，还是时间换空间？ #

## Spark 存储系统用于存储 3 个方面的数据 ##

**Spark 存储系统用于存储 3 个方面的数据，分别是 RDD 缓存、Shuffle 中间文件、广播变量。**

RDD 缓存、Shuffle 中间文件、广播变量，这 3 个服务对象是 Spark 应用性能调优的有力“抓手”，而它们又和存储系统有着密切的联系，因此想要有效运用这 3 个方面的调优技巧，我们就必须要对存储系统有足够的理解。

1.  RDD 缓存：

RDD 缓存指的是将 RDD 以缓存的形式物化到内存或磁盘的过程。对于一些计算成本和访问频率都比较高的 RDD 来说，缓存有两个好处：一是通过截断 DAG，可以降低失败重试的计算开销；二是通过对缓存内容的访问，可以有效减少从头计算的次数，从整体上提升作业端到端的执行性能。

1.  Shuffle 中间文件

在很多场景中，Shuffle 都扮演着性能瓶颈的角色，解决掉 Shuffle 引入的问题之后，执行性能往往能有立竿见影的提升。因此，凡是与 Shuffle 有关的环节，你都需要格外地重视。

Shuffle 的计算过程分为 2 个阶段：

*  Map 阶段：Shuffle writer 按照 Reducer 的分区规则将中间数据写入本地磁盘；
 *  Reduce 阶段：Shuffle reader 从各个节点下载数据分片，并根据需要进行聚合计算。

Shuffle 中间文件实际上就是 Shuffle Map 阶段的输出结果，这些结果会以文件的形式暂存于本地磁盘。在 Shuffle Reduce 阶段，Reducer 通过网络拉取这些中间文件用于聚合计算，如求和、计数等。**在集群范围内，Reducer 想要拉取属于自己的那部分中间数据，就必须要知道这些数据都存储在哪些节点，以及什么位置。而这些关键的元信息，正是由 Spark 存储系统保存并维护的。**

1.  广播变量
2.

在日常开发中，广播变量往往用于在集群范围内分发访问频率较高的小数据。\*\*利用存储系统，广播变量可以在 Executors 进程范畴内保存全量数据。\*\*这样一来，对于同一 Executors 内的所有计算任务，应用就能够以 Process local 的本地性级别，来共享广播变量中携带的全量数据了。

## Spark存储系统的基本组件 ##

### 什么组件存储什么内容 ###

与调度系统类似，Spark 存储系统是一个囊括了众多组件的复合系统，如 BlockManager、BlockManagerMaster、MemoryStore、DiskStore 和 DiskBlockManager 等等。

1.  BlockManager  
    不过，家有千口、主事一人，**BlockManager 是其中最为重要的组件，它在 Executors 端负责统一管理和协调数据的本地存取与跨节点传输。**

*  对外，BlockManager 与 Driver 端的 BlockManagerMaster 通信，不仅定期向 BlockManagerMaster 汇报本地数据元信息，还会不定时按需拉取全局数据存储状态。另外，不同 Executors 的 BlockManager 之间也会以 Server/Client 模式跨节点推送和拉取数据块。
 *  对内，BlockManager 通过组合存储系统内部组件的功能来实现数据的存与取、收与发。

1.  MemoryStore 和 DiskStore

Spark 存储系统提供了两种存储抽象：MemoryStore 和 DiskStore。BlockManager 正是利用它们来分别管理数据在内存和磁盘中的存取。

广播变量的全量数据存储在 Executors 进程中，因此它由 MemoryStore 管理。Shuffle 中间文件往往会落盘到本地节点，所以这些文件的落盘和访问就要经由 DiskStore。相比之下，RDD 缓存会稍微复杂一些，由于 RDD 缓存支持内存缓存和磁盘缓存两种模式，因此我们要视情况而定，缓存在内存中的数据会封装到 MemoryStore，缓存在磁盘上的数据则交由 DiskStore 管理。

### 以什么格式存储 ###

有了 MemoryStore 和 DiskStore，我们暂时解决了数据“存在哪儿”的问题。但是，这些数据该以“什么形式”存储到 MemoryStore 和 DiskStore 呢？\*\*对于数据的存储形式，Spark 存储系统支持两种类型：对象值（Object Values）和字节数组（Byte Array）。\*\*它们之间可以相互转换，其中，**对象值压缩为字节数组的过程叫做序列化，而字节数组还原成原始对象值的过程就叫做反序列化。**

对象值(读写快占空间)和字节数组(读写慢省空间)二者之间存在着一种博弈关系，两者之间该如何取舍，我们还是要看具体的应用场景。**核心原则就是：如果想省地儿，你可以优先考虑字节数组；如果想以最快的速度访问对象，还是对象值更直接一些。** 不过，这种选择的烦恼只存在于 MemoryStore 之中，而 DiskStore 只能存储序列化后的字节数组，毕竟，凡是落盘的东西，都需要先进行序列化。

### 透过 RDD 缓存看 MemoryStore ###

在 RDD 的语境下，我们往往用数据分片（Partitions/Splits）来表示一份分布式数据，但在存储系统的语境下，我们经常会用数据块（Blocks）来表示数据存储的基本单元。**在逻辑关系上，RDD 的数据分片与存储系统的 Block 一一对应，也就是说一个 RDD 数据分片会被物化成一个内存或磁盘上的 Block。**

这些包含 RDD 数据值的 MemoryEntry 和与之对应的 BlockId，会被一起存入 Key 为 BlockId、Value 是 MemoryEntry 引用的链式哈希字典中。

总的来说，RDD 数据分片、Block 和 MemoryEntry 三者之间是一一对应的，当所有的 RDD 数据分片都物化为 MemoryEntry，并且所有的（Block ID, MemoryEntry）对都记录到 LinkedHashMap 字典之后，RDD 就完成了数据缓存到内存的过程。

#### 内存空间不足 ####

如果内存空间不足以容纳整个 RDD 怎么办？”很简单，强行把大 RDD 塞进有限的内存空间肯定不是明智之举，所以 Spark 会按照 LRU 策略逐一清除字典中最近、最久未使用的 Block，以及其对应的 MemoryEntry。相比频繁的展开、物化、换页所带来的性能开销，缓存下来的部分数据对于 RDD 高效访问的贡献可以说微乎其微。

### 透过 Shuffle 看 DiskStore ###

相比 MemoryStore，DiskStore 就相对简单很多，因为它并不需要那么多的中间数据结构才能完成数据的存取。\*\*DiskStore 中数据的存取本质上就是字节序列与磁盘文件之间的转换，\*\*它通过 putBytes 方法把字节序列存入磁盘文件，再通过 getBytes 方法将文件内容转换为数据块。

# 7.内存管理基础：Spark如何高效利用有限的内存空间？ #

## 内存的管理模式 ##

在管理方式上，**Spark 会区分堆内内存（On-heap Memory）和堆外内存（Off-heap Memory）**。这里的“堆”指的是 JVM Heap，因此堆内内存实际上就是 Executor JVM 的堆内存；堆外内存指的是通过 Java Unsafe API，像 C++ 那样直接从操作系统中申请和释放内存空间。

*  堆内内存的申请与释放统一由 JVM 管理。
 *  堆外内存则不同，Spark 通过调用 Unsafe 的 allocateMemory 和 freeMemory 方法直接在操作系统内存中申请、释放内存空间

## 内存区域的划分 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 10]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 11]

1.  堆外内存  
    Spark 把堆外内存划分为两块区域（执行与缓存内存）：

*  一块用于执行分布式任务，如 Shuffle、Sort 和 Aggregate 等操作，这部分内存叫做 **Execution Memory**；
 *  一块用于缓存 RDD 和广播变量等数据，它被称为 **Storage Memory**。

1.  堆内内存

*  堆内内存的划分方式和堆外差不多，Spark 也会划分出用于执行和缓存的两份内存空间。
 *  不仅如此，Spark 在堆内还会划分出一片叫做 User Memory 的内存空间，它用于存储开发者自定义数据结构。
 *  除此之外，Spark 在堆内还会预留出一小部分内存空间，叫做 Reserved Memory，它被用来存储各种 Spark 内部对象，例如存储系统中的 BlockManager、DiskBlockManager 等等。

**我们在前三块内存的利用率上有比较大的发挥空间，因为业务应用主要消耗的就是它们，也即 Execution memory、Storage memory 和 User memory。而预留内存我们却动不得，因为这块内存仅服务于 Spark 内部对象，业务应用不会染指。**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 12]

## 执行与缓存内存 ##

*  执行内存： 用于执行分布式任务，如 Shuffle、Sort 和 Aggregate 等操作，这部分内存叫做 **Execution Memory**；
 *  缓存内存： 用于缓存 RDD 和广播变量等数据，它被称为 **Storage Memory**。

在所有的内存区域中，最重要的无疑是缓存内存()Storage Memory)和执行内存(Execution Memory)，而内存计算的两层含义也就是数据集缓存和 Stage 内的流水线计算，对应的就是 Storage Memory 和 Execution Memory。

**执行任务主要分为两类**：

*  一类是 Shuffle Map 阶段的数据转换、映射、排序、聚合、归并等操作；
 *  另一类是 Shuffle Reduce 阶段的数据排序和聚合操作。它们所涉及的数据结构，都需要消耗执行内存。

在 Spark 1.6 版本之前，Execution Memory 和 Storage Memory 内存区域的空间划分是静态的，一旦空间划分完毕，不同内存区域的用途就固定了。也就是说，即便你没有缓存任何 RDD 或是广播变量，Storage Memory 区域的空闲内存也不能用来执行 Shuffle 中的映射、排序或聚合等操作，因此宝贵的内存资源就被这么白白地浪费掉了。

考虑到静态内存划分潜在的空间浪费，在 1.6 版本之后，Spark 推出了统一内存管理模式。统一内存管理指的是 Execution Memory 和 Storage Memory 之间可以相互转化，尽管两个区域由配置项 spark.memory.storageFraction 划定了初始大小，但在运行时，结合任务负载的实际情况，Storage Memory 区域可能被用于任务执行（如 Shuffle），Execution Memory 区域也有可能存储 RDD 缓存。

**执行任务相比缓存任务，在内存抢占上有着更高的优先级。**

Execution Memory 和 Storage Memory 之间的抢占规则，一共可以总结为 3 条：

*  如果对方的内存空间有空闲，双方就都可以抢占；
 *  对于 RDD 缓存任务抢占的执行内存，当执行任务有内存需要时，RDD 缓存任务必须立即归还抢占的内存，涉及的 RDD 缓存数据要么落盘、要么清除；
 *  对于分布式计算任务抢占的 Storage Memory 内存空间，即便 RDD 缓存任务有收回内存的需要，也要等到任务执行完毕才能释放。

### 结合代码看内存消耗 ###

val dict: List[String] = List(“spark”, “scala”)
    val words: RDD[String] = sparkContext.textFile(“~/words.csv”)
    val keywords: RDD[String] = words.filter(word => dict.contains(word))
    keywords.cache
    keywords.count
    keywords.map((_, 1)).reduceByKey(_ + _).collect

![在这里插入图片描述][20210601160848562.png]

首先，

*  第一行定义了 dict 字典，这个字典在 Driver 端生成，它在后续的 RDD 调用中会随着任务一起分发到 Executor 端。
 *  第二行读取 words.csv 文件并生成 RDD words。
 *  第三行很关键(User Memory )，用 dict 字典对 words 进行过滤，此时 dict 已分发到 Executor 端，Executor 将其存储在堆内存中，用于对 words 数据分片中的字符串进行过滤。Dict 字典属于开发者自定义数据结构，因此，Executor 将其存储在 User Memory 区域。
 *  第四行(Storage Memory )和第五行用 cache 和 count 对 keywords RDD 进行缓存，以备后续频繁访问，分布式数据集的缓存占用的正是 Storage Memory 内存区域。
 *  第六行(Execution Memory): 在 keywords 上调用 reduceByKey 对单词分别计数。reduceByKey 算子会引入 Shuffle，而 Shuffle 过程中所涉及的内部数据结构，如映射、排序、聚合等操作所仰仗的 Buffer、Array 和 HashMap，都会消耗 Execution Memory 区域中的内存。

ache 之后 再进行count，主要是因为cache不是action算子，所以需要一个action算子来触发缓存的生效。count是为了trigger cache的计算，计算cache的过程中，在展开（Unroll）之前，会消耗Execution memory；展开之后，转化为MemoryEntry，消耗的就是Storage memory。

### 问题 ###

1.  你知道启用 off-heap（开启对外内存spark.memory.offHeap.enabled=true) 之后，Spark 有哪些计算环节可以利用到堆外内存？你能列举出一些例子吗？
2.  相比堆内内存，为什么在堆外内存中，Spark 对于内存占用量的预估更准确？  
    堆内内存：因为spark只是将无用的对象引用删除，但是无用对象真正的回收还要依赖于JVM来管理。Spark只是做了标记，但是真正什么时候删除spark并不知道，这里存在一个时间差。  
    相比较堆外内存：spark自己做管理就可以清楚的知道当前还有多少内存空间可以使用。
3.  堆外内存和堆内内存的分配计算方法：  
    Executor的内存主要分为三块：第一块是让task执行我们自己编写的代码时使用，默认是占Executor总内存的20%；第二块是让task通过shuffle过程拉取了上一个stage的task的输出后，进行聚合等操作时使用，默认也是占Executor总内存的20%；第三块是让RDD持久化时使用，默认占Executor总内存的60%。

spark.memory.offHeap.enabled， 默认为false ,如果为 true，Spark 将尝试使用堆外内存进行某些操作。如果启用了堆外内存使用，则spark.memory.offHeap.size必须为正。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 13]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 14]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 15]  
结合我在上面给定的配置参数，，计算不同内存区域（Reserved、User、Execution、Storage）的具体大小

**堆内内存中：**  
保留内存300M，  
用户内存为 20*0.2=4GB，  
Storage内存为 20*0.8*0.6=9.6GB，  
Execution内存为 20*0.8\*0.4=6.4GB

**堆外内存中：**  
Storage内存为 10*0.6=6G，  
Execution内存为 10*0.4=4G

### spark官方建议谨慎使用堆外内存 ###

原因其实很简单，在于堆外堆内的空间互不share，也就是说，你的task最开始用堆外，用着用着发现不够了，这个时候即使堆内还有空闲，task也没法用，所以照样会oom。

内存本来就有限，再强行划分出两块隔离的区域，其实反而增加了管理难度。tungsten在堆内其实也用内存页管理内存（Tungsten的相关优化，可以参考后面Tungsten那一讲），也用压缩的二进制数据结构，因此gc效率往往可以保障，这也是为什么官方推荐就用堆内就可以了。

堆外内存默认不会开启，不开启就不会用。只有开启off heap，spark才会尝试用堆外，否则全部用堆内。实际上，spark社区推荐用堆内，因为tungsten的优化机制，良好的数据结构，会把gc的开销降到最低，堆内也能很好地完成计算

如果开辟了堆外，spark会优先用堆外，直到用光了为止，再去用堆内。所以如果堆外空间不充分，执行任务对缓存空间的抢占会更严重，oom的风险也越高。

我不觉得有什么场景一定要用堆外，就我看来，对于开发者来说，堆外更多地是一种备选项，是Optional的。不过，尽管如此，我们还是要知道堆外、堆内各自有哪些优缺点、优劣势，这样在结合应用场景做选择的时候，也能有的放矢~

开启堆外之后，

*  执行任务(execution memory)默认会走堆外，堆外用尽了，后续的任务才会走堆内。
 *  对于缓存(storage memory)来说，如果你明确指定了用off heap，那就是明确走堆外，如果你不明确指定，那么默认走堆内。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20221014/e14cdb7a1d7f42a8b5b48693f2a58dc1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]: /images/20221014/5e7d7b841f864b73880497ab8e1665b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]: /images/20221014/9668ceed5863427c8100d12a87ed272e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]: /images/20221014/e25ae0e84dbd4fd98bb0165bd6c92bcb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 4]: /images/20221014/3d61ba71e7904ee19c80483f5cd34596.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 5]: /images/20221014/d67c822b958a4d21b605cac1f1c52288.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 6]: /images/20221014/54aade33001b45c7be40e708dd9e4e59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 7]: /images/20221014/9d6677595ce541ddb4e84dd8908c763b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 8]: /images/20221014/b3c74a7d27b94a96aadf63078cbd0a32.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 9]: /images/20221014/830aac0bc5ea4c91ba437041a283525a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 10]: /images/20221014/d6d9da5c04be42039864bf7e0be908b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 11]: /images/20221014/2553cc0e925b4aee9bee7bf07b188ee0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 12]: /images/20221014/7ba40372b8f34ed49b7cc0f802254ac6.png
[20210601160848562.png]: /images/20221014/a92f71599f144ac99aeeaa40055d7bc9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 13]: /images/20221014/080b8ed891164b2588d80bc7eaebd5af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 14]: /images/20221014/4e13ca4188d64f549068860af551716f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 15]: /images/20221014/a447aa0b342a4002a32bc2589281d27d.png