Spark最佳实践

前言

本文主要分为四个部分：

1. 分布式计算概览：第一章将会从基础的 分布式计算理论 开始，讨论一个分布式计算系统需要实现哪些 主要的功能，以及业界通用的解决方案，并在最后简单扩展了下分布式计算系统的发展历程。这部分主要为开发人员 奠定分布式计算系统的重要理论概念。
2. Spark技术细节：基于第一章讨论的理论知识，在第二章中我们将会深入讨论Spark是如何 通过从逻辑执行图转化为物理执行图 来实现分布式计算的。随后着重讨论了 Shuffle过程与管理、内存模块、数据共享 等其他模块细节，建立系统的Spark知识体系。
3. Spark性能优化：基于前两章的基础知识的铺垫，本章将会从 程序开发、资源配置、数据倾斜、Shuffle管理、内存使用 等方面提供优化技巧，可以 为开发人员提供相关的调优思路。
4. Spark最佳实践：在最后一章中，作者整理了大量Spark开发过程中的 实践经验与应用技巧，从 编码、监控、数据处理 等几个大维度提供Spark应用程序开发的最佳实践思路参考。

---

一、分布式计算概览

1.1 基本概念

分布式计算 即一个计算过程将会在多台机器上进行。

组件之间彼此进行交互以实现一个共同的目标：把需要进行大量计算的工程数据分区成小块，由多台计算机分别计算，再上传运算结果后，将结果统一合并得出数据结论。

与之相反的是 集中式计算，一个运算过程全都在一台机器上执行，比较典型的就是MySQL这种单机数据库。

简单说就是 100个人干活和1个人干活的区别。

分布式计算概念其实很早就有了，并不是因为进入数据爆发时代之后才催生的。
就像CPU从单核变多核一样，然后发展出超线程这种技术。

从单台机器的集中式计算发展为多台机器的分布式计算是随着计算机的发展自然而然出现的。

分布式计算是一门计算机科学的研究课题，涉及到许多分支技术（CS模型、集群技术、通用型分布式计算环境等）。

本文讨论的主要内容是：从分布式计算的理论基础中实现，并且已经得到了大规模生产环境验证的计算框架。

1.2 如何实现

我们先抛开具体的计算框架或者执行引擎，思考一下如果要我们自己实现一个分布式计算的系统，你会如何做？

考虑这么一个简单的应用场景：

某后端应用为了提高负载，分别在10台独立的服务器上部署了服务。服务对应的日志信息都打到各个服务器中的一个统一路径下，现要求 统计访问量前10的IP与其对应的访问次数（PV）。

同样为了简单起见，这里假设日志信息都是格式化的数据，以逗号作为分隔符的csv形式。

最直观的方式就是将各个服务器上的日志数据导入一个MySQL实例中，然后跑一下统计的SQL就完事儿了：

select ip,count(1) as cnt from log_table group by ip;

这种方式是不是很熟悉？就是开篇提到的 集中式计算，先将数据集中到一起之后执行运算程序。

但是，现在问题来了。

如果访问量暴涨，各个服务器上的日志数据大到单个机器 需要运行很久甚至跑不出结果 该如何处理？

你可能会说换硬件，提升CPU性能和核心数。

这确实是一个办法，但并不是一个一劳永逸的办法。因为单机能够提升的空间总是 有限度的 ，不可能无限升级扩张，而且高配服务器价值不菲。

从另外一个角度想，虽然单机无法做到无限升级，但是 机器数量是可以无限的进行横向扩展的！ 跑不过来了，加机器！

好，既然有了解决思路，那么我们又弄来了一台服务器准备开整。

但是好像还有点问题，即使现在多了一台机器，但是MySQL并 不知道要分多少数据给新来的老铁。

就算我们人为55开将数据对半分出去，再手动到两台机器上跑相同的SQL。

仍然不能直观的得到我们想要的结果，因为两个MySQL跑出来的数据 都是不完整的，我们需要再 把两部分结果聚合起来才能得到最终的结果。

总结一下，我们缺少这样一个软件，它能够：

• 帮助我们拆分计算逻辑
• 将拆分后的计算逻辑分发到各个可用节点上
• 协调各个节点的计算能力执行任务
• 将各个节点的结算结果汇总整合

这就是一个分布式计算系统需要实现的 基本功能。

一个分布式计算框架就是一个很好地解决了以上问题的软件。

现在我们再来重新捋一捋，要实现一个具备上述功能的软件，我们都需要重点关心哪些问题。

1.3 数据在哪儿

集中式计算的特点是 移动数据、统一计算，就像我们之前将所有数据移动到一个MySQL实例中计算一样。

与之相反的，分布式计算很重要的一个特点就是：移动计算而不移动数据。

他所崇尚的是，将计算逻辑（即代码片）分发到数据所在位置上执行。相对于庞大的数据来说，一个代码片的大小显得非常的微不足道，传输消耗可以忽略不计。

也就是说要实现分布式计算，第一个要解决的问题就是 数据在哪儿。

这也就是为什么分布式计算系统经常与分布式存储系统一起使用的原因，因为分布式存储系统可以很好的解决 数据在哪儿 这个问题。

常见的分布式存储系统应用海量数据的分发方案：

• 1.数据量分布
• 2.数据范围分布
• 3.Hash分片/消息队列
• 4.一致性Hash

本质上他们所要实现的目的是一样的： 将数据尽可能的分散在各个节点中，并可以提供高效的使用。

这些内容属于另外一个系统话题了，有兴趣的朋友可以自己去了解一下HDFS、HBase、Cassandra等分布式存储系统的实现，这里就不过多的涉及了。

回到之前的案例，现在你可能会觉得把数据都移动到一个MySQL实例这种办法很蠢也很耗时。

数据不已经在服务器上了么，不已经解决 数据在哪儿 这个问题了吗，可以直接用命令行一个个执行统计出结果来：

hosts=(host1 host2 host3 ...) for host in ${hosts[*]} do # 假设IP在第三列，使用awk实现分组统计的功能 ssh user@$host "cat xxx.log | awk -F , '{a[$3]+=1}END{for (i in a){printf "%s,%d\n",i,a[i]} }'" done

这不挺好的么，不用移动数据，连接到目标机器后执行计算逻辑获得结果。

在当前这个案例中，如果后端应用请求能够负载均衡到各个节点上，那么我们就可以得到一个天然的、均匀分布在各个节点上数据集。再通过一个简单的shell脚本实现“分布式”计算的这么一个机制。

虽然现在我们可以通过这个脚本来遍历各个服务器并执行计算，但是 本质上计算所花费的时间和集中式计算所花费时间是一个级别的（因为他是串行的），而且我们仍然不能够直接得到最终想要的结果。

如果计算逻辑更复杂一些，那么这个简单的机制将会很快崩溃，因为我们根本不知道聚合之后的逻辑如何处理。

我们再总结归纳一下，在解决了 数据在哪儿 这个问题之后，我们要来实现分布式计算首先要解决其中两个最重要的问题：

• 1.如何拆分计算逻辑
• 2.如何分发计算逻辑

1.4 计算逻辑处理

1.4.1 拆分逻辑

计算逻辑要实现分布式，就必须要解决：如何将一个巨大的问题拆分成相对独立的子问题分发到各个机器上求解。

这个和动态规划分解问题、自底向上求解的思想有些类似。

从 在哪里发生计算 的角度来看，所有的计算逻辑都能够划分为这两种类型：

• 1.能够分发到各个节点上并行执行的
• 2.需要经过一定量的结果合并之后才能继续执行的

在上面统计PV的例子中，从日志中解析出IP 这个行为就是属于 能够分发到各个节点上并行执行的 ，各个节点间互不影响。

而 根据IP进行分组统计 这个行为则 需要经过一定量的结果合并之后才能够继续执行 ，如：要计算127.0.0.1这个IP的访问次数，则要等待该IP的所有数据都在一起之后才能够统计出正确的结果。

1.4.2 分发逻辑

计算逻辑拆分的问题之后，面临的将是如何将拆开的逻辑分发出去。

在上面的例子中通过脚本连接到远程机器执行计算逻辑已经解决了一部分的任务分发问题。

对于 能够分发到各个节点上并行执行的 的任务来说：

• 任务的个数：取决于数据块的个数
• 任务的执行位置：取决于数据块所在的位置

这里将是分布式计算和集中式计算最大的不同点，移动计算逻辑而不移动数据。

这样，我们就简单地对这个计算逻辑完成了分解与拆分，但是这仅仅是一个为了方便说明而举的例子。

真正的计算逻辑分发还需要解决很多问题。

例如，对于 需要经过一定量的结果合并之后才能够继续执行 的任务来说，如何决定任务的个数与执行任务的位置则并不能这么直观的解决：

• 聚合任务的个数和执行位置如何决定？
• 聚合结果中，各个节点的中间结果往哪存、怎么存？
• 聚合节点什么时间通过什么方式获取所需要的数据？
• 网络延迟、中断这种情况如何处理？

两者之间需要建立一种可靠的通讯机制以保证准确无误地完成计算任务。为了使整个求解过程完美的衔接起来，你需要解决一系列的 通信、容灾、任务调度等问题。

1.5 MapReduce

在现实生活中，产生的问题和其对应的解法是千变万化的，需要有一个抽象的框架来概括这些问题和解法。

由于自己动手撸一个分布式计算系统需要解决的问题和任务实在是太多了，而现实生活中又那么迫切的需要有这么一个系统可以处理大规模的数据。

首先对此公开提出解决方案的是Google的MapReduce。

其将一个分布式任务定义为两种类型的Job组成：

• 1.Map Job对应的就是可以在各个节点上一起执行互不影响的逻辑，即能够分发到各个节点上并行执行的 的任务
• 2.Reduce Job处理的则是Map产生的中间结果（如果有），即需要经过一定量的结果合并之后才能够继续执行 的任务

Map和Reduce之间通过一个Shuffle过程来链接。 组成一个完整的分布式计算流程所有的细节问题都将在这个过程中得到有效的处理。

在先前的例子中，从日志中解析出IP 为Map Job，根据IP进行分组统计 为Reduce Job。

reduce通过网络Shuffle到拉取map端属于自己的数据（某个IP的所有数据）之后，进行聚合统计最后输出一个统一的结果。

如上图所示，一个MapReduce任务会经历以下阶段：

• 将Map任务分发到数据块所在的各个节点上执行

• Map任务读取数据块，并执行相关的计算逻辑

  • 对于Map类型任务的分发其实可以归为分布式存储系统的问题
  • 基本上所有分布式计算框架都是基于优先数据本地化进行的
  • 也就是说数据存在哪，计算就分发到哪

• Map任务的结果将会写入内存的环形缓冲区中

  • 根据key和对应的partition算法对数据进行分区，相同key的数据位于同一个分区中
  • 经过分组、排序等流程操作之后，不同内存块中相同分区的数据写入磁盘中

• 启动Reduce任务，连接各个Map任务节点，从磁盘中读取自己需要处理的分区数据

  • Reduce任务的个数可手动设置也可根据分区个数来
  • 相同分区的数据必定在一个Reduce中，一个Reduce任务可能处理多个分区的数据
• 执行Reduce中的业务逻辑并输出

以上就是分布式计算中MapReduce的思想，现在的分布式计算引擎基本上都是基于MapReduce实现的，从细节上看计算模型可能不是MapReduce，但是宏观的计算思想上看都是属于MapReduce的一种实现，所以也被称为类MapReduce计算框架。

现在，我们有了MapReduce这个工具之后，再来实现之前的PV统计就十分简单了：只需要在Map函数接口中定义日志解析规则，在Reduce函数接口中定义分组聚合逻辑，剩下的事情框架会帮你都处理好并输出一个最后的结果。

用户再也不用担心 计算逻辑如何拆分、拆分完如何分发 这些底层的问题（MapReduce具体的实现细节篇幅原因本文不做阐释，感兴趣的朋友自行查阅资料），只需要专注于业务逻辑实现即可。

1.7 发展历程

当前热门的计算引擎都能归类到Hadoop的生态圈中，是08年以来一直都很热门的技术栈。

04年前后，Google发布了《The Google File System》、《MapReduce: Simplified Data Processing on Large Clusters 》和《Bigtable: A Distributed Storage System for Structured Data》，Nutch的作者看了之后用Java实现了Hadoop的三个核心组件：HDFS、MapReduce和HBase分别对应三篇论文阐述的理论思想。

Hadoop最开始是作为Nutch的一个子项目存在的，解决的问题就是搜索引擎对应的大规模数据存储、计算，直到被Yahoo孵化为一个完整的分布式系统解决方案并在08年之后全球广泛应用。

Hadoop作为一个分布式系统，也是由两部分组成的：分布式存储和分布式计算。

MapReduce对应的就是最初Hadoop分布式计算的解决方案。

Hadoop发展到现在，可以集成在其中的计算框架非常多，从应用的角度讲大体上可以分为两种：离线和实时。
有些开发者将这些计算引擎按照出生年代、计算模型归了个类：

第一代：无实时计算能力

MapReduce

第二代：基于MapReduce之上的工具与实时计算能力

Tez：基于MapReduce的DAG执行工具。

Storm：以拓扑结构的方式提供实时计算能力。

第三代：完整的技术栈与更加稳定的系统

Spark：提供更高效、更完整的技术栈。

Heron：比Storm更加稳定的实时系统。

第四代：更为先进的计算思想与框架的统一

Flink：最接近DataFlow数据流思想的解决方案。

Beam：多种计算框架的兼容与支持。

---

二、Spark技术细节

总体上来说，Spark的流程和MapReduce的思想很类似，只是实现的细节方面会有很多差异。

有了以上分布式计算的基本概念之后，我们再来具体看Spark里面的细节就会比较清晰明了。

首先澄清2个容易被混淆的概念：

• 1.Spark是基于内存计算的框架
• 2.Spark比Hadoop快100倍

第一个问题是个伪命题。

任何程序都需要通过内存来执行，不论是单机程序还是分布式程序。

Spark会被称为 基于内存计算的框架 ，主要原因在于其和之前的分布式计算框架很大不同的一点是，Shuffle的数据集不需要通过读写磁盘来进行交换，而是直接通过内存交换数据得到。效率比读写磁盘的MapReduce高上好多倍，所以很多人称之为 基于内存的计算框架，其实更应该称为 基于内存进行数据交换的计算框架。

至于第二个问题，有同学说，Spark官网 就是这么介绍的呀，Spark run workloads 100x faster than Hadoop。

这点没什么问题，但是请注意官网用来比较的 workload 是 Logistic regresstion。

注意到了吗，这是一个需要反复迭代计算的机器学习算法，Spark是非常擅长在这种需要反复迭代计算的场景中（见问题1），而Hadoop MapReduce每次迭代都需要读写一次HDFS。以己之长，击人之短 差距可向而知。

如果都只是跑一个简单的过滤场景的 workload，那么性能差距不会有这么多，总体上是一个级别的耗时。

所以千万不要在任何场景中都说 Spark是基于内存的计算、Spark比Hadoop快100倍，这都是不严谨的说法。

2.1 逻辑执行图

2.1.1 弹性分布式数据集

RDD是Spark中的核心概念，直译过来叫做 弹性分布式数据集。

所有的RDD要么是从外部数据源创建的，要么是从其他RDD转换过来的。RDD有两种产生方式：

• 从外部数据源中创建
• 从一个RDD中转换而来

你可以把它当做一个List，但是这个List里面的元素是分布在不同机器上的，对List的所有操作都将被分发到不同的机器上执行。

RDD就是我们需要操作的数据集，并解决了 数据在哪儿 这个问题。

有了数据之后，我们需要定义在数据集上的操作（即业务逻辑）。

回想一下我们之前经历的流程：

• 1、一开始我们什么都没有，只有分散在各个服务器上的日志数据，并且通过一个简单的脚本遍历连接服务器，执行相关的统计逻辑
• 2、我们接触了MapReduce计算框架，并定义了Map和Reduce的函数接口来实现计算逻辑，从而用户不比关心计算逻辑拆分与分发等底层问题

虽然MapReduce已经解决了我们分布式计算的需求，但是其编程范式只有map和reduce两个接口，使用不灵活。

在Spark中，RDD提供了比MapReduce编程模型丰富得多的编程接口，如：filter、map、groupBy等都可以直接调用实现（这些操作本质上也划分为Map和Reduce两种类型）。

现在，统计PV的例子中实现计算逻辑的伪代码可以这么写：

// 从外部数据源中创建RDD，即读取日志数据 val rdd = sc.textFile("...") // 解析日志中的ip rdd.map(...) // 根据ip分组 .groupBy("ip") // 根据分组结果统计数量 .map(x=> (x._1, x._2.size)) // 保存到外部数据源 .saveAsTextFile("...")

在RDD进行操作行为可以划分为两种：

• Transformation：如filter、map、groupBy等，将会产生另外一个RDD
• Action：如count、saveAsTextFile等，触发整个逻辑图的计算流程

一个Spark程序可以看做是 一个或者多个RDD的完整生命周期，从诞生到发展，到变换，再到输出之后销毁。

2.1.2 依赖关系

现在你可能会问，使用MapReduce，通过指定数据源定义了操作数据集，通过Map和Reduce两个函数接口划分了 能够分发到各个节点上并行执行的 和 需要经过一定量的结果合并之后才能够继续执行的 两种任务，并基于这两种接口类型的任务去拆分和分发计算逻辑。

那么Spark中是如何做的呢？

Spark中通过RDD定义了 分布式数据集，通过RDD的编程接口定义了计算逻辑，但是Spark是如何根据RDD中定义的逻辑来划分 能够分发到各个节点上并行执行的 和 需要经过一定量的结果合并之后才能够继续执行的 任务，从而实现计算逻辑的拆分和分发呢？

其实和MapReduce一样，Spark中虽然提供了丰富的算子给用户实现计算逻辑，但是这些算子最终仍然会被归为两类：Map和Reduce。

前面我们说过，在RDD上执行Transformation操作会产生另外一个RDD，随即，RDD之间将会产生依赖关系和父子RDD关系。

而RDD中的依赖关系分为两种：

• 1、完全依赖：又称为窄依赖，父RDD中一个分区的数据只被子RDD中对应的一个分区使用（1对1）
• 2、部分依赖：又称为宽依赖，父RDD中一个分区的数据会被子RDD中对应的多个分区使用（1对多 or 多对多）

看到了吗，最后RDD通过依赖关系又回到了我们之前讨论的话题：能够分发到各个节点上并行执行的 和 需要经过一定量的结果合并之后才能够继续执行的 两种任务。

对于完全依赖来说，各个分区之间的任务是互不影响的，所以其能够发到各个节点上并行执行。

对于部分依赖来说，子RDD的某个分区可能依赖于父RDD的多个分区，所以其需要经过一定量的结果合并（依赖的所有父RDD分区）之后才能够继续执行。

定义了这两种类型的任务之后，Spark就可以根据依赖关系进行 计算逻辑的拆分与分发 。

那么RDD上的哪些操作是宽依赖，哪些操作是窄依赖呢？

其实仔细想想很好区分，对于map、filter这种不需要Shuffle的操作都是窄依赖，而groupBy、reduceBy等需要Shuffle聚合的操作都是宽依赖。

通过Transformation操作，RDD之间将会产生依赖关系，基于RDD上的操作与依赖关系 将会形成一张逻辑执行图 来描述本次任务的计算过程。

通过Action操作触发该逻辑执行图的计算流程。

什么意思呢？

在RDD上进行的Transformation操作都是惰性执行的，意思就是只有数据真正用到的时候（Action操作）才会进行Transformation操作。

例如以下RDD的操作：

//rdd1只保留了从rdd中计算而来的路径，没有真正执行计算 val rdd1 = rdd.map.map.filter.map //直到有action操作才会触发计算任务 rdd1.count

也就是说，count之前，我们写的计算逻辑其实只是在 画一个逻辑图，只有真正使用到了count的时候，整个逻辑图才会被触发并执行计算逻辑。

这么做的原因要归咎到RDD的计算模型，当rdd中出现action操作的时候，spark将会生成一个job，并根据rdd的依赖关系画出一张逻辑执行图。

费劲心机画出了逻辑图之后再划分物理图时将会有最关键的作用。

2.2 物理执行图

从RDD上得到逻辑执行图之后，执行计算任务前期的准备工作就都完成了，现在我们来详细讨论一下Spark是如何 拆分、分发计算逻辑的。

Spark将会划分逻辑图从而生成物理执行图，表现形式为 DAG有向无环图，RDD的执行模型将根据物理图的划分而展开。

现在我们知道，基于逻辑执行图，由于RDD之间的依赖关系被明显的划分为了两种：

• 对于完全依赖，可以完全不管其他RDD或者其他分区的执行进度，直接一条走到底的
• 对于部分依赖，需要父RDD不同分区中的数据，所以他 一定是等到所有父RDD计算完毕之后才会执行的

基于逻辑执行图和对应的依赖关系，Spark可以明显的 划分出Stage：

• 从逻辑图的最后方开始创建Stage
• 遇到完全依赖则加入当前Stage
• 遇到部分依赖则新建一个Stage

由此对整个逻辑图进行Stage的划分。这就是Spark对于计算逻辑的组织和拆分方式。

那么这么做有什么好处呢？

基于Stage的独立性，Spark实现了 Pipeline的计算方式。且由于 Stage内部的操作只有完全依赖，它可以毫无顾忌的建立 回溯机制：当一个分区数据计算失败或者丢失，可以直接从父RDD对应的分区中恢复，而不是重新计算整个父RDD。

如果所有操作都是立即执行的话，那么处理流程应该是这样子的：

//读取数据 list1 = readAllFromHDFS //将所有数据进行对应的map转换操作 list2 = list1.map //将所有数据进行对应的map转换操作 list3 = list2.map //将所有数据进行对应的filter过滤操作 list4 = list3.filter

注意，这种模式下，每个步骤都需要将 全量的数据集加载到内存中操作 这是毋庸置疑的，每个操作都要等待前一个操作全部处理完毕。

作为对比，我们再来看Pipeline的计算模式：

//读取数据 data = readOneLineFromHDFS //读取一条处理一条，每条数据经过管道执行到末端 data.map.map.filter

数据是作为流一条条从管道的开始一路走到结束，每个Stage都是一条独立的管道。最为直观的好处就是：不需要加载全量数据集，上一次的计算结果可以马上丢弃。

全量数据集其实是一个很恐怖的东西，全世界都在避免它。所以某种意义上来看，如果没有Shuffle过程，Spark所需要内存其实非常小，一条数据又能占多大空间。

第二，如果不是Pipeline的方式，而是马上触发全量操作，势必需要一个中间容器来保存结果，其实这里就又回到MapReduce的老路，效率很低。

现在我们来考虑 不根据RDD的依赖关系来划分Stage的前提下，两种比较极端的情况：

• 1、整个逻辑图作为一个Stage

  • 一个Job只包含一个Stage，数据一路从头走到尾，什么中间结果都不需要保存
  • 如果RDD之间都是 完全依赖 的话这是最完美的场景

  • 缺陷：

    • 在Shuffle操作符处（即部分依赖的产生处），只能通过一个Task来处理所有分区的数据
    • 多个Task情况下没有办法各自感知Shuffle过程中所需要的数据状态
    • 严重影响计算效率

• 2、每个RDD的操作都作为一个Stage

  • 各个操作都需要进行全量计算，其实就相当于MapReduce
  • 缺陷：严重影响计算效率

可以看到，Spark通过RDD之间的依赖关系来划分逻辑执行图形成一个个独立的Stage，并通过Stage来实现Pipeline的计算模式。

计算逻辑拆分后，通过Pipeline的执行将计算逻辑分发到各个节点，并最大程度保证计算的效率。

综上，基于逻辑执行图能做的事情有：

• 1、划分Stage
• 2、执行Pipeline
• 3、建立回溯机制

根据RDD之间的依赖关系来划分Stage解决了以下问题：

• 1、实现Pipeline，不需要保留中间计算结果
• 2、计算保持高效，Task分布均衡

至此，Spark主要的 计算逻辑拆分与分发 步骤大概介绍完毕。

与之相对的，一段Spark代码，或者一个Spark程序，运行起来之后是什么样子的，代码是如何被调度执行的，应该在开发的阶段就能在脑子里形成一个执行图。

充分了解程序运行的背后发生了什么是保证系统稳定高效运行的关键，这点放在哪里都是真理。

2.3 Shuffle过程与管理

2.3.1 Shuffle总览

和之前看到的MapReduce Shuffle过程相对比。二者在高级别上来看别没有多大区别，都是将mapper中的数据进行partition之后送到不同的reducer中，reducer以内存为缓存边拉取数据边计算。

但是在具体实现的低级别角度上两者区别还是比较大的，MapReduce阶段划分明显，Spark中没有明显的划分。

MapReduce中的Mapper即为Spark中的ShuffleMapTask，而Reduce对应的可能是ShuffleMapTask或者ResultTask。

Spark各个阶段通过RDD的算子体现出来，具体Shuffle过程可以分为：

• 1、Shuffle Write
• 2、Shuffle Read

Write过程其实很简单，根据之前划分的Stage，每个Stage的final task的结果将会写磁盘，和MapReduce一样，有多少个分区数就会写多少个文件。

后续的Stage将会通过网络来fatch各自对应的数据文件。

Read过程需要解决几个问题：

• 1、什么时候fetch数据：依赖的stage中所有ShuffleMapTask都执行完之后才进行fetch，迎合pipeline的思想
• 2、如何获得数据位置：ShuffleMapTask结束之后都会想Driver端汇报数据存放位置，ResultTaskfetch数据时都会向Driver查询需要fetch的数据在哪里，Driver端有比较复杂的实现机制
• 3、fetch的数据怎么存：刚fetch过来的数据存放在softBuffer中，计算之后的数据可以根据策略选择存放在内存或者内存+磁盘中

和fetch过程的计算和MapReduce也不一样：

• 1、Spark：边fetch边计算，因为是无序的，所有没有必要要求所有数据都获取之后才进行计算
• 2、MapReduce：MapReduce中强制要求数据有序之后才进行reduce操作，所以MapReduce是 一次性fetch所有数据之后才计算

总结一下，与MapReduce相比：

• Height Level：无太大区别，将mapper中的数据进行partition之后送到不同的reducer中
• Low Level：实现差别较大，MapReduce阶段划分明显，Spark中没有明显的划分

Spark 中负责 Shuffle 过程管理的是 ShuffleManager，它接管了 Shuffle Read、Shuffle Write 过程中的 执行、计算和处理 相关的实现细节。

比如 Write 过程中怎么组织数据写入磁盘，Read 过程中怎么拉取数据和保存数据。

ShuffleManager 是一个接口，主要有两种实现：

• HashShuffleManager：Spark 1.2之前默认使用，会产生大量的中间磁盘文件，进而由大量的磁盘IO操作影响了性能。
• SortShuffleManager：每个Task在进行Shuffle操作时，虽然也会产生较多的临时磁盘文件，但是最后会将所有的临时文件合并（merge）成一个磁盘文件，因此每个Task就只有一个磁盘文件。在下一个stage的Shuffle read task拉取自己的数据时，只要根据索引读取每个磁盘文件中的部分数据即可。

2.3.3 HashShuffleManager

为了简单的说明，这里假设我们的 Executor 可用的CPU核心数只有一个，无论 Executor 上有分配了多少个Task，同一时间只能执行一个Task。

在 Shuffle Write 阶段，Executor 的在依次执行每个Task时，HashShuffleManager 都会对Task 中的所有数据的key执行相应的hash运算，hash的参数是下游Stage的
Task数量。通过这hash映射之后，每个key都会有一个对应的结果值，根据hash的结果值来写文件（这个过程中会经过一段内存的缓冲区，缓冲区满了之后写入磁盘），相同结果的数据写到一个文件中。

这样一来，上游每个Task中，都会根据下游Stage的Task数量 产生对应数量的文件，相同key的数据肯定在同一份文件中，一份文件中可能会有多个key的数据。下游stage计算数据时，只需要拉取这个文件的所有数据即可进行计算。在这个过程中，会边拉取边计算，每个Task也会有自己的缓冲区，每次只取buffer大小的数据通过内存中的Map进行聚合，反复操作直至数据获取完毕。

这么描述大家可能还会觉得合情合理，那么我们从产生的总文件数的角度来看呢？

假设当前Stage有200个Task需要执行，下游Stage有100个Task，按照我们刚刚描述的过程来看，产生的总文件数为：200 * 100 = 20000 个。

这是一个非常惊人的数字，我们都知道 磁盘的IO 一直是程序执行的瓶颈之一，我们在执行程序的时候都会尽可能的避免写磁盘操作。而现在，一个 Shuffle Write 过程就会产生成千上万个文件，注定了这个程序不会快到哪里去。

工作流程如下图所示：

那么有没有优化方式呢？肯定是有的。

在使用 HashShuffleManager 的时候，我们可以设置一个参数 spark.Shuffle.consolidateFiles 该参数默认值为false，将其设置为true即可开启优化机制。墙裂建议设置为true，为啥呢？我们来解释解释。

consolidate机制最重要的功能就是 允许不同的task复用同一批磁盘文件，这样就可以有效将多个task的磁盘文件进行一定程度上的合并，从而大幅度减少磁盘文件的数量，进而提升Shuffle write的性能。

之前的流程中，每个Task都会创建n个文件，Task之间是互相隔离的。而在 consolidate机制 中Task之间是可以复用文件的，因为同一个key的数据可能是分布在不同的Task上处理的。

简单来说，因为Task之间的数据文件可以复用，一个cpu核心只会创建和下游Stage的Task数量一样多的文件数，同一个cpu核心中处理的所有task都会重复使用同一批文件。

• consolidate=false：文件数量由 上游Stage任务数（不同的任务可能会被同一个cpu处理） * 下游Stage任务数 决定
• consolidate=true：文件数量由 上游处理任务的CPU核心数（一个cpu可能会处理多个任务） * 下游Stage任务数 决定

还是我们之前举的例子，假设当前Stage有200个Task需要执行，下游Stage有100个Task，如果此时我们有10个Executor（每个1core），那么总文件数为： 10（cpu核心数）* 100（下游Task数量） = 200 个。

由此可见，当开启了 consolidate 机制后，Executor 的cpu核心数越多，在提供处理并行度的同时，Shuffle Write 产生的文件数就越多，这点需要注意。

Shuffle Read 阶段并没有变化，都是直接拉取自己所需要的那份数据进行计算。

consolidate机制下，工作流程如下：

2.3.4 SortShuffleManager

经过前面的介绍之后，我们知道使用 HashShuffleManager 时开启consolidate机制可以减少很多文件的产生，提高 Shuffle Write 效率。

无论 consolidate机制 是否开启，HashShuffleManager 所产生的文件数都与下游Stage的Task数量有关系。

现在我们再来看另外一种 Shuffle管理机制，SortShuffleManager。

通过 SortShuffleManager 这个名字大家可以知道，这是一个排序的Shuffle管理器（HashShuffleManager为无序）。

Shuffle Write 的具体执行过程如下：

• 每个Task将Shuffle的数据写入自己的buffer内存缓冲区中，每条数据写入时都会判断是否超出阈值

• 超出使用阈值则触发刷写，将数据一批批的写入磁盘中

  • 写入磁盘前会根据key对内存中的数据进行排序
  • 排完序后的数据根据批次大小（默认10000）依次写入磁盘中

• Task数据处理结束后，将之前刷写的所有文件读取，合并之后重新写入一个大文件中

  • 因为一个Task处理的数据可能对应下游多个Task需要处理的数据
  • 所以此过程会创建索引文件标记下游各个Task对应的数据在文件中的start offset与end offset
• 由于需要标记下游各个Task所需要的数据偏移量，所以需要进行sort排序之后才可写入

从以上过程中可以看出，和 HashShuffleManager 一样 SortShuffleManager 的每个Task也会创建很多文件，不同的是 HashShuffleManager 中每个Task创建的文件数和下游的Stage任务数一致，而 SortShuffleManager 则是 按照自己的buffer内存空间大小刷写的文件快，并且最后还会做一次大合并，一个Task只对应一个文件。

文件数量由 上游Stage的Task数量 决定。

执行流程如下：

除此之外，SortShuffleManager 还有另外一种 bypass 的执行模式。

当 Shuffle map task数量小于 spark.Shuffle.sort.bypassMergeThreshold 的值，且不是聚合类的Shuffle算子（比如reduceByKey），比如 join 等操作时将会触发。

此时task会为每个下游task都创建一个临时磁盘文件，并将数据按key进行hash然后根据key的hash值，将key写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的。

最后，同样会将所有临时磁盘文件都合并成一个磁盘文件，并 创建一个单独的索引文件。

该过程的磁盘写机制其实跟未经优化的HashShuffleManager是一模一样的，因为都要创建数量惊人的磁盘文件，只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件，也让该机制相对未经优化的HashShuffleManager来说，Shuffle read的性能会更好。

而该机制与普通SortShuffleManager运行机制的不同在于：第一，磁盘写机制不同；第二，不会进行排序。

也就是说，启用该机制的最大好处在于，Shuffle write过程中，不需要进行数据的排序操作，也就节省掉了这部分的性能开销。

2.4 内存模块

Spark是用Scala开发完成的，也是一个运行在JVM体系上的系统性框架，所以 Spark的内存模型也是基于Java虚拟机来的。

基本模型就是：堆、栈、静态代码块和全局空间，在虚拟机的内存模型上Spark将内存做了二次划分。

作为一个严重依赖内存进行数据计算的系统来说，内存管理模块 是Spark中极其重要的一部分。

2.4.1 内存划分

从性质上看，Executor 可使用的内存空间分为两种：堆内、堆外。

堆内内存即直接通过 spark.executor.memory 或者 -–executor-memory 设置分配得到，属于一定会有值的强制性配置。

而堆外内存则是一种可选性配置，默认不使用，通过配置 spark.memory.offHeap.enabled 参数启用，由 spark.memory.offHeap.size 参数设定堆外空间的大小。

对外内存将会 存储经过序列化的二进制数据。一定程度上会减少不必要的内存开销，以及频繁的 GC 扫描和回收，提升了处理性能。

从内存区域上看，内存大致可以划分为三个模块（堆外内存没有 Execution 的空间）：

• 1、Storage：RDD缓存、Broadcast等数据空间。
• 2、Execution：Shuffle过程使用的内存。
• 3、Other：用户定义的数据结构、Spark内部元数据等其他内存空间。

2.4.2 内存管理

这里主要讨论堆内内存的管理。

静态内存管理

静态内存管理为 Spark 1.6 之前默认使用的管理方式。

程序运行时Storage、Execution和Other三块内存大小是固定的：

• 1、Storage：默认60%，由 spark.storage.memoryFraction 控制
• 2、Execution：默认20%，由 spark.Shuffle.memoryFraction 控制
• 3、Other：默认20%，等于 1 - spark.storage.memoryFraction - spark.Shuffle.memoryFraction

此外，Storage 和 Execution 区域中分别存在着 安全阈值 的设置（spark.storage.safetyFraction 与 spark.Shuffle.safetyFraction，默认值80%），防止内存超出使用范围而造成的OOM异常。

在静态内存管理的方式中，用户实际使用时 需要根据数据集本身的特性和分布情况 来调整三个内存模块各自占据的大小。

统一内存管理

在内存结构总体不变的情况下，Spark 1.6 之后引入了新的内存管理机制，统一内存管理。

和静态内存管理相比，统一内存管理主要的变化点在于：

• 增加系统预留的内存空间
• 各个区域初始分配的默认值重新调整
• Storage和Execution两个区域之间不再是固定大小，而是 动态调节

总堆内内存的基础上，先扣除 给系统的预留空间（默认300M），剩下的为可用内存总大小。

在可用内存总大小的基础上，划分两大块：

• 统一内存（Unified Memory）：Storage、Execution共同使用的内存，默认值 0.75（2.0以后为0.6），由 spark.memory.fraction 控制

  • Storage：默认为0.5，占统一内存的50%，由 spark.storage.storageFraction 控制
  • Execution：默认为0.5，占统一内存的50%，等于 1 - spark.storage.storageFraction
• 其他内存（Other）：默认为0.4，占可用总内存的40%， 等于 1 - spark.memory.fraction

可以看到，在统一内存管理中，Storage和Execution被划入一块大内存池中进行统一管理。

这样做的好处是，Storage和Execution 的内存空间用户可以不用自己那么操心去优化、调整。

当有一方的内存不够用时，将会到另外一方去「借」一些内存回来用，达到 动态内存分配与调整 的效果。

在 Spark 1.6 之后的版本中默认不再使用 静态内存管理 的方式，但是可以通过设置 spark.memory.userLegacyMode 的值（true/false，默认false）来选择内存管理方式。

2.6 数据共享

2.6.1 广播变量

Spark通过广播变量为用户提供了一种 以极低代价获取全局一致的数据信息 的途径。

考虑这么一个场景：我们在编写Spark应用程序的时候，经常会定义一个 本地变量，可能是一个List、一个Map等等。在后续RDD的转换操作过程中，我们会经常引用到这个本地变量做一些逻辑处理。

这种情况下，由于RDD的操作是在 各个节点上分布式执行的，但是本地变量的定义一般在Driver端。

为了使各个节点上的Task能够读取到这个变量，Spark会在 各个Task中创建这个变量并复制数据。这意味着一份普通的本地变量在使用的时候可能会被复制 成千上万遍（各个节点上会有n个Task执行）。

这对于网络和内存资源来说是很大的一种浪费，如果本地变量稍微大一些，那么Executor的内存将很快耗尽导致任务失败。

但是从Executor的角度上看，其实它只需要 一份数据给这个节点上的所有Task共享使用 就可以了。

由于现实场景中有很多诸如以上的情况出现，Spark引入了 广播变量 来解决这个问题。

Spark将会把广播变量 复制一份保存到各个Executor节点上：

• 1、各个节点中对该数据的所有引用，可以 直接访问本地数据进行计算而不需要通过网络获取。
• 2、避免了在每个task中都创建一份数据对象造成的内存资源浪费。

广播变量的应用范围非常广，可以让用户完美的避开类似 大表 join 小表 这种场景中消耗大量的网络与内存资源。

广播变量细节

知道广播变量是什么以及能做到什么时候，我们来讨论一下广播变量的实现以及一些使用限制。

在代码中通过 SparkContext 创建 Broadcast 变量时，在 Driver 端将会把该数据写入本地文件夹中：

//创建广播变量 val bcst = sc.broadcast(xxxTable) //在rdd的转换操作过程中使用该广播变量的值 rdd.filter(r => bcst.value.contains(r))

bcst 这个广播变量的信息会在 submit task 的时候 随着计算逻辑一起被序列化发送到各个节点上，此时各个节点上只有一份广播变量信息，并没有真正的数据。

节点上数据初始化的过程如下：

• 当有 action 操作触发 job 执行的时候，各个 Executor 将会开始反序列化计算逻辑执行任务。
• 反序列化过程中，如果计算逻辑中引用了广播变量会到本地的 BlockManager 查找数据，本地不存在的话则根据广播变量的信息到 Driver端拉取广播变量的数据。
• 通过网络拉取到数据之后交给 BlockManager 写入本地存储，后续的所有数据引用都直接读取本地数据。

在这个过程中，需要重点关注的是：Executor 通过什么方式到 Driver 获取数据呢？

最简单的做法就是在Driver端启动一个Http服务监听某个端口，等待Executor连接到此端口后传输数据。看起来简单高效，Spark就是通过 HttpBroadcast 来实现这个功能。

这种方式在 广播变量比较小、Executor数量比较少 的情况下很适用，因为Driver不会产生什么负担。

但是如果广播变量稍微大一点，Executor的数量稍微多一些，可以想象Driver会承受多少 网络数据传输的压力，一旦Driver出现瓶颈可能会导致整个任务的失败。

所以需要有一个更加聪明的方法来解决这个问题。

其实解决办法在我们的生活中很常见，大家想一想 迅雷的BT下载 是怎么回事儿就知道了。

除了 HttpBroadcast 之外，Spark还提供了 TorrentBroadcast 来实现 BT下载 的功能，过程如下：

• Driver

  • 将数据序列化为字节数组，并按照 block size 切分成块
  • 将分块数据写入 BlockManager，并由 BlockManager 负责记录 哪个数据块被哪个节点保存了
  • 启动Http Server监听端口

• Executor

  • 随机获取 广播变量的分块信息，查询本地 BlockManager 是否有该数据块，没有则到Driver查询该数据块信息

  • 开始BT过程：

    • 在 Driver 的 BlockManager 获取拥有数据块分片 的 Executor
    • 如果数据分片在 Executor 上，则请求对应的节点，否则继续从 Driver 中获取数据
    • 将获取到到的数据块写入本地 BlockManager 中
    • 通知 Driver 的 BlockManager 当前节点存储的数据块信息
    • 所有的数据块信息都拿到诸侯，合并本地的数据块，并反序列化存入BlockManager
    • 删除BlockManager之前fetch的数据副本

通过BT下载，客户端下载越多，能够提供数据服务的节点就越多，Driver永远不会因为广播变量的数据传输产生瓶颈。

至此，广播变量的过程细节大致结束，但是从上面对广播变量的介绍来看，我们也可以看到一些比较明显的缺陷和限制：

• 不可广播大数据集：数据集太大仍然会对 集群网络、节点内存 造成压力，使用时切记评估数据量
• 广播变量不可变：一旦数据被广播数据，那么它就不可能被继续修改（想想为什么？），如果要修改只能 更新本地变量之后重新广播一遍

2.6.2 RDD缓存

了解完广播变量之后，我们再来看另外一个可以 提供全局共享数据 的操作，RDD缓存。这也是 Spark 相对比 MapReduce 最大不同的一点：通过RDD的缓存实现中间数据复用。

广播变量能够实现 小数据集 在各个节点间的 本地化共享，而RDD缓存则能够实现 大数据集 的 全局共享。

前面我们说过，MapReduce每个task都会 不断写磁盘且无法重用和容错，而 Spark通过 Pipeline 尽量避免持久化操作，并通过 Cache、Checkpoint等操作提供RDD重用。

考虑这么一个场景：在对RDD进行了1234567个操作之后得到一个需要被 反复使用的中间RDD。在不对这个RDD进行缓存的情况下，每次引用这个RDD做后续计算时，都会查找这个RDD的血缘关系，并从源头RDD开始重新计算（还记得Pipeline吗？）。

在这样的场景中，如果我们能够将中间RDD缓存起来，后续的所有引用和计算都可以直接拿到结果开始执行 岂不是妙哉？

Spark 为RDD提供了两种操作实现数据复用：

• Cache

  • 缓存位置：内存
  • 使用范围：本次 Application，一旦Driver进程结束就会将数据清空

• Checkpoint

  • 缓存位置：磁盘
  • 使用范围：任何Application都可以读取使用，永久持久化，除非手动移除

使用过程非常简单，只需要在RDD上调用 cache/checkpoint 接口即可，缓存的动作将会在下一个 action 操作中提交执行。

RDD缓存完后，后续所有涉及到该RDD的计算都将直接从内存中读取分区数据。

Cache操作的实现

action 触发 job 之后，在计算 RDD的partition时，会调用 rdd.iterator() 判断该RDD是否要被cache，如果需要，则RDD的所有分区在各自的节点上都会经过以下过程：

• 调用 CacheManager 生成 RDDBlockId

• 根据该id到 BlockManager 查找 是否有 checkpoint

  • 有，则直接读取 checkpoint 的数据
  • 没有，则执行计算逻辑，得到结果
• 将得到的数据交给 BlockManager 进行缓存

各个节点的 BlockManager 缓存完各自的分区数据后将会 告知 Driver 数据缓存情况，至此 RDD已被缓存到各个节点的内存中。

如何读取已缓存的RDD呢，流程如下：

• 使用到 cached rdd 时，task先调用本地 BlockManager.getLocal() 从memoryStore中读取

• 如果没有cache在本地，则通过 BlockManager.getRemote() 读取其他节点上cached数据

  • 通过 Driver 可拿到该 RDD 缓存的全局位置信息

不同的缓存级别

• MEMORY_ONLY: 默认缓存策略，相当于cache操作。直接将数据以Java对象格式存储与内存中，如果内存空间不足那么剩下的数据将不会被缓存。未被缓存的数据被使用到时将会根据RDD的lineage重新计算。

  • 优点：速度最快
  • 缺陷：内存消耗高
  • 适用：相对于数据量，内存空间足够大的情况

• MEMORY_AND_DISK: 存储格式同 MEMORY_ONLY ，当内存空不足时将会把数据写入磁盘。

  • 优点：使用磁盘作为内存的补充
  • 缺陷：写入磁盘的部分将会消耗比较多的时间
  • 适用：内存缓存空间不足的情况

• MEMORY_ONLY_SER: 存储策略同 MEMORY_ONLY，存储格式为序列化后的字节流

  • 优点：更加节省内存的使用并避免频繁的GC
  • 缺陷：会带来序列化和反序列化的开销
  • 适用：内存空间可能不足且不想因为写磁盘而消耗太多性能
• MEMORY_AND_DISK_SER: 存储策略同 MEMORY_AND_DISK ，不同的是存储格式为序列化后的字节流
• DISK_ONLY: 使用未序列化的Java对象格式，将数据全部写入磁盘文件中。
• 对于上述任意一种持久化策略，如果加上后缀_2，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上，如MEMORY_ONLY_2、MEMORY_AND_DISK_2, 等等。主要作用是容错，避免某个节点挂了之后缓存的数据无法找到。

开发编码过程中，需要根据 可用硬件资源、RDD数据量、程序时效性等要求 选择不同的缓存策略。

虽然看起来缓存这个功能很美好，但是在实际使用过程中，RDD的缓存使用有利有弊。

例如在内存资源不足的情况下，如果还强行将RDD缓存到内存中，可能会导致原本多花点时间就可以执行成功的任务因为 内存爆满而失败。即使将RDD缓存的级别调整成内存+磁盘甚至全磁盘的形式，仍然可能 会导致任务进展缓慢，Executor节点失去响应等情况，所以使用RDD缓存时需要根据数据量与资源情况进行调整优化。

---

三、Spark性能优化

经过1、2章基础理论知识的讨论，我们现在可以具备了一定的Spark开发能力。了解了Spark是什么以及能够做到什么，接下来我们将会讨论实际使用场景中，如何对Spark程序进行性能优化以及一些使用技巧。

3.1 开发调优

代码开发，是执行Spark任务的第一步，同时也是优化Spark任务的第一个入手点，良好的 RDD lineage、高性能的算子操作、不同高级特性的组合使用，都能够给Spark任务带来巨大的提升空间。

开发出优秀的Spark程序，需要你熟悉Spark的各种API和特性。其中最重要的一点我们在逻辑执行图小节中提到过：开发Spark程序其实就是在画图。

如何能够把这个图快速画出来的同时还能画好看，就是你需要考虑的，这就是考验Spark开发的基本功。

3.1.1 业务逻辑梳理

无论是开发Spark应用还是其他任何程序，开发人员对于业务逻辑的了解和掌握程序从一开始就决定了这个程序的好坏。

正确的理解业务逻辑，并能够 将其中复杂的处理过程梳理清楚、拆分明确，这对于后续的编码过程将起到决定性的作用。

试想一下，如果你不了解要开发的逻辑是什么，只会生搬硬套做API的搬运工，怎么会知道代码中的哪个地方可以用 更加优雅和高效的代码代替和优化 呢？

所以对业务逻辑的梳理是开发编码的第一步，也是最基础的一步，下面所说的开发优化点都是基于开发人员对业务有深刻理解的基础上的，否则即使看了这些文字性的描述回去也无法在你自己的程序上找到应用的地方，看了白看。

不要着急做功能实现，功能实现很简单。

难的是怎么以最好、最快、最高的运行效率执行无误。

部门、业务线甚至整个公司的数据计算需求一定是源源不断而且越来越复杂的，意味着平台上跑的计算任务将会越来越多且越来越复杂。

如果一直都是直接生搬硬套翻译实现业务逻辑，用不了过久平台就会被计算任务压垮。

所以最明智也是最优雅的选择是，一开始就在业务逻辑上下重功夫，保证计算程序从一开始就是健康、稳定的。

3.1.2 RDD复用

和其他任何程序中 变量复用 一样，在Spark程序中创建并使用RDD也要贯彻这个思想。

在编码的时候，RDD和任何单机程序一样，本身只是作为一个普通的变量对象存在，不同的是单机变量的创建会消耗内存，而RDD的创建会 消耗磁盘、内存与算力等更多方面的资源（想想RDD创建之后的使用过程，是不是这样呢）。

所以要把RDD的创建和使用当做一个 需要消耗高昂费用的动作 来谨慎使用，从代码的源头节约与优化程序空间与效率。

有的同学在开发Spark程序的时候，可能在业务逻辑1创建了一个RDD1，经过各种Transformation以及最后的Action操作之后，开始处理业务逻辑2，又在相同的数据源上创建了RDD2，然后继续写业务逻辑代码。

一般来讲，相同的数据源的RDD 只允许创建一次，不要创建相同的RDD，保证代码的整洁性。

在RDD的lineage过程中，如果有多个业务重复使用某个lineage的计算过程，则 应该将其抽出作为一个独立的中间RDD使用，尽可能复用相同的RDD。

无论是数据源RDD还是中间RDD，如果被反复多次使用，则应该考虑将其做 缓存持久化操作。

可以看到，如果没有对业务逻辑有比较清晰的了解，开发人员很难从繁杂的计算过程中提取出可以复用甚至进行缓存操作的代码块，无法优化到点。

另外，在考虑对RDD持久化操作时，应该针对 可用硬件资源、RDD数据量、程序时效性等要求 选择不同的缓存策略（详见「内存模块」小节）。

总结：

• 相同的数据源的RDD 只允许创建一次
• 多个业务逻辑反复使用同一个lineage 应该将其抽出作为一个独立的中间RDD使用
• 任何被多次重复使用的RDD应该考虑将其做 缓存持久化操作

3.1.2 算子优化

RDD复用之后，你就拥有了一个高效、强大的数据源，但是仅仅拥有数据源是不够的，如何对数据源进行操作才是后面的重头戏。

很多开发人员在开发过程中仅仅充当了一个「翻译员」的角色，用Spark API将自然语言的业务逻辑「翻译」成Spark程序。

这么做虽然也能实现功能，完成需求，但是就和前面说的一样，功能实现很简单，难的是怎么以最好、最快、最高的运行效率执行无误。

这才是你对于平台的意义所在。

基于你对自身业务需求的理解，我们要求做到以下几点：

1、尽量避免使用Shuffle类算子

通过「Shuffle过程与管理」这一小节中的介绍，我们知道Spark程序中Shuffle过程是 最消耗性能的部分，会产生大量的网络IO、磁盘IO等资源消耗，能避免就尽量避免。

这是影响任务执行的最大因素，如果在源头不通过Shuffle就能够实现业务逻辑，一定不要贪图方便直接调用Shuffle算子实现。

虽然可能会多写一点代码，但这对于你、对于程序来说都是百利的。

如果能在源头就消除Shuffle过程，那也就不存在之后产生的数据倾斜、Shuffle调优等过程了。

2、Shuffle无法避免的情况下，使用带有预聚合功能的Shuffle操作

如果你的业务逻辑必须通过Shuffle才能实现，那么 务必选择性能与性价比最高的算子。

比如分组统计功能，简单点就是直接使用 groupBy，但是还有更具性价比的算子，比如 reduceBy、aggregateBy。

这两个算子将会在各个Executor本地做一次 预聚合（类似 MapReduce 的 Combiner），在各个节点中，相同key的数据会按照同样的业务逻辑先聚合一次，所以在进行Shuffle的时候数据量会减少很多，Shuffle过程带来的影响也就越小。

3、使用高性能的算子

除了第二点提到的使用预聚合的算子代替直接Shuffle的算子之外，其他算子也有对应的「高性能」版本。

如mapPartitions替代普通map、foreachPartitions替代foreach等，其效率高的原因是partition版本的操作针对的是一个partition中的所有数据，而普通的map或者foreach是在所有分区中一条条遍历的，批量操作与单条操作的效率区别。

比如说一个任务有100个分区，每个分区1w条数据。在写数据库的时候用foreachPartitions同时操作100个分区，批量写入100w数据，而foreach虽然也是同时操作100个分区，但是每个分区同时写入的数据只有1条，也就是一次写入100条，差距异常明显。

但是有的时候，使用mapPartitions会出现OOM（内存溢出）的问题。因为单次函数调用就要处理掉一个partition所有的数据，如果内存不够，垃圾回收时是无法回收掉太多对象的，很可能出现OOM异常。

所以使用这类操作时也要慎重！

3.1.3 使用共享变量

Spark中提供的共享变量是一个提升程序效率的强大工具，所以我们应该想尽办法在程序中高效的用上他们。

以广播变量为例，正确的使用广播变量能够让开发人员 回避许多原本需要Shuffle算子才能完成的操作，也就是上一小节中我们提到的 「尽量避免使用Shuffle类算子」。

假设我们现在有RDD1和RDD2、对应的业务逻辑如下：

// RDD1数据量10w // RDD2数据量10000w // 求RDD2中不包含RDD1的数据 // 实现方式1，使用join操作 rdd2.join(rdd1).filter(去除关联不到的部分) // 实现方式2，使用广播变量 val bst = sc.broadcast(rdd1.collect.map(转换为kv形式的格式)) rdd2.filter(bst.value.contains)

以上两种方式实现的效果是一样的，但是执行效率天差地别，共享变量就是Spark为开发人员提供的这么一种「神兵利器」，一些维表、本地变量、join的小表等等都可以通过广播变量包装之后使用以提升程序效率。

3.1.4 序列化优化

Spark应用中，Broadcast、Shuffle、Cache等地方都需要用到序列化，如何保证Java对象高效的序列化与反序列化也是程序高效执行的重要因素。

Spark中默认使用Java的序列化框架，但是其同时也支持Kryo作为底层序列化框架使用，因为Kryo比Java自身的序列化框架性能更加强大（官方介绍，Kryo序列化机制比Java序列化机制，性能高10倍左右），因此我们推荐都使用Kryo：

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

除了底层序列化框架的选择之外，程序中我们也可以为序列化/反序列化过程的优化“添房加瓦”。

具体怎么做呢？-- 对程序中使用的数据结构进行优化。

JVM中有三种比较消耗内存的类型：对象、字符串和集合。

• 每个对象除了携带业务数据之外，还有对象头、引用、对象大小等元数据信息。
• 每个字符串内部都有一个字符数组以及长度等额外信息。
• 集合类型内部通常会使用一些内部类来封装集合元素，比如Map.Entry

因此我们建议，编码过程中，能用值类型就不用引用类型。

比如Int、Long代替String、String代替Object、Array代替List。尽量减少内存使用与序列化/反序列化开销，降低GC频率提升提升性能。

但是在实际编码过程中，这个规则很难执行到位，因为在「面向对象」的编码思想下，「一切皆为对象」。所有代码都使用基本数据类型，那么项目的可读性和可维护性会很差，因此需要根据项目情况酌情挑选数据结构使用。

3.2 资源调优

现在你已经写了一个 优雅且高效的Spark程序，接下来你会将其部署到集群上执行，在执行之前你需要设置一系列的资源配置来使你的程序有足够的资源和环境来执行。

很多开发人员在使用 spark-submit 的时候可能不知道需要设置哪些参数以及如何设置，最后可能就是从别的程序脚本抄一个或者胡乱给个数字甚至不设置。

资源配置过程很简单，但也很重要。

给多了浪费，给少了运行慢，多了少了可能都会出现异常情况。

所以给多少资源也是一门学问，我们需要对Spark资源分配使用的原理有清晰的理解，尽力避免给大量资源无脑执行也不关心资源分配的情况。

在使用spark-submit时，以下参数 必须设置且必须仔细评估程序之后设置：

• num-executors：当前Spark作业共需要多少个Executor节点来执行，不设置则使用默认的数量1。

  • 根据当前作业读取的 数据源数量、大小与分区情况 进行设置
  • 分区数量多可以酌情多给几个节点提升并行度，分区数量少给再多也是浪费可以视情况减少几个节点

• executor-memory：Executor可使用内存的大小，num-executors * executor-memory 即为Executor需要的总内存大小。

  • 这是能够直接决定你的程序 能否执行以及能够快速执行 的重要参数，给少了OOM常相见
  • 通常每个Executor给的内存大小4-8G，但是只是个建议值，具体还是得看 数据源分区的大小以及程序中的Shuffle、Cache、Broadcast等操作所需要的内存（内存划分见「内存模块」小节）。

• executor-cores：Executor可使用的CPU核心数，num-executors * executor-cores 即为当前程序需要的总CPU核心数。

  • 由于Executor并不能很好的发挥多核的计算能力，因此在「少Executor 多核」与「多Executor 少核」两种情况下应该更倾向于后者

• driver-memory：Driver可使用内存的大小，driver-memory + num-executors * executor-memory 即为当前程序需要的总内存大小。

  • Driver一般情况下1G足矣，但是如果代码中有collect大量数据的话需要增大Driver内存，否则OOM见。

以下参数属于「高级选项」，可以视情况调整：

• spark.default.parallelism：stage的默认task数量。

  • 第一个stage的task数量往往 依赖于数据源的分区数，没有显式repartition等操作设置分区数的话，Shuffle之后第二个stage的task数量默认取决于这个参数
  • 如果没有显式设置的话，默认情况下task数量会比较少，不利于任务并行度的提高。
  • 官方建议设置为 Executor可用总CPU核心数的2-3倍，以充分利用Executor的计算资源最大化并行度。

• spark.storage.memoryFraction：Spark1.6以前的版本适用，控制内存中有多少空间用来持久化RDD，默认为60%。

  • 如果程序中需要缓存的RDD数据较多且你希望数据都缓存在内存中，那么可以适当调高这个值，以免内存不够时写入磁盘影响效率
  • 如果缓存的RDD数据不多，可以降低这个值，多提供些内存给代码执行和Shuffle使用，提升效率

• spark.Shuffle.memoryFraction：Spark1.6以前的版本适用，控制内存中有多少空间用来执行Shuffle过程，默认为20%。

  • 如果程序中有大量的Shuffle操作，内存的使用超出了20%的限制，那么将会产生 disk spill 刷写到磁盘，大大影响执行效率。此时可以适当增加该参数的值。

• spark.memory.fraction：Spark1.6之后的版本适用，控制内存中有多少空间来缓存RDD与执行Shuffle过程，默认75%（2.0之后为60%）。

  • 此参数控制RDD缓存与Shuffle执行的总内存，如果内存空间不够的情况下可以调高此参数。

• spark.storage.storageFraction：Spark1.6之后的版本适用，控制 spark.memory.fraction 中有多少内存给RDD缓存使用，默认50%。

  • 1.6之后的统一内存管理中，storage和Shuffle两边的内存是可以自动动态调整的，一般情况下不需要人为干预
  • 如果需要显示减小storage的比例的话可以调低此参数，Shuffle可用的内存比例为 1 - spark.storage.storageFraction

3.3 数据倾斜

代码写好了，程序跑的资源也经过精心调配之后设置好了，没有其他意外情况的话，你的Spark程序已经能够正常的跑在集群上。

但是别以为这样就结束了，这仅仅是Spark程序生命周期的开始。

为了给你的程序保驾护航，你还需要时刻关注 新上线的应用程序的执行情况是否健康、是否如你所愿如你所想。

应用程序的执行情况你都可以在 Spark或者Yarn的WebUI 上查看到，有非常详细的执行信息。

我们现在来讨论一个 可能是导致程序执行缓慢甚至异常 的最大罪魁祸首： 数据倾斜。

什么是数据倾斜呢？

就是 绝大多数数据（比如80%以上甚至更多）都被分配到 绝少数的节点 上执行（比如20%甚至更少）。

这么一来，意味着剩下绝大多数的节点都没处理或者没怎么处理数据，处于空闲状态。而 少数节点则一直处理非常忙碌的状态，任务处理需要排队，节点完成计算任务耗时非常长，其他完成任务的节点就在旁边看热闹，但是 只有等所有节点都完成了计算任务整个程序才能算完成。

例如，总共有1000个task，990个task都在10分钟之内执行完了，但是剩余10个task却要三、四个小时，整个程序的执行时间由最长的那个task决定（反过来的木桶效应）。

同时，因为某些节点上处理的数据量太多，根据不同的业务代码操作，可能还会出现某些节点在Shuffle过程或者数据处理过程出现OOM异常导致程序失败。

简单来讲，就是 几颗老鼠屎坏了一锅粥。

导致数据倾斜的原因有很多，但是其本质都是一样的：在Shuffle等需要通过网络读写数据的过程中，因为数据key分布不均匀，导致大部分数据被集中获取到少部分节点上。

数据倾斜的情况可以在WebUI上的Stages、Executors页面中观察到，这也就是为什么我们要求对于初次上线的应用，需要时刻关注新上线的应用程序的执行情况是否健康、是否如你所愿如你所想。

在Web界面中，有哪些Stage，Stage中有哪些Task，各个Task处理的数据量和执行计算的时间等等，这些你都可以很清晰的看到。

如果发现你的应用中有存在有 几个Task处理的数据量明显比其他Task要大很多，而且还在不停的处理数据，而其他Task已经执行完毕，那么你就是遇到了数据倾斜的问题。

那么如果我们确定了程序中存在数据倾斜的情况，该如何处理呢？

根据数据倾斜产生的原因，我们可以在 不同的切入点使用不同的处理策略。

3.3.1 定位代码与数据问题

在Web界面上，我们可以直观的获得 发生数据倾斜的Stage对应的代码行数，但这个行数并不能精准直接定位到发生数据倾斜的代码，因为它显示的是当前Stage开始执行的代码行数。

由于数据倾斜只有可能在Shuffle过程中发生，所以 导致数据倾斜的一定是会产生Shuffle过程的算子，比如groupByKey、reduceByKey、aggregateByKey、join、distinct、repartition等等。

所以，你只需要在Stage所在的行数向上查找Shuffle操作符，那么其就是导致数据倾斜的罪魁祸首。

找到问题代码之后，需要做的事情很明显了吧？优化之。

此时我们需要统计一下该Shuffle操作符所使用的数据源，观察各个数据源的 key分布情况（如每个key有多少数据量），以及导致数据倾斜的key在哪里、都有哪些。

根据数据情况与你对业务的理解，使用「开发调优」中算子优化提到的技巧，尽量这个Shuffle操作的影响降到最低。

3.3.2 处理源头数据

如果该Shuffle操作符无法避免，代码层面上无法做太多优化，那么此时可以考虑 预先处理数据源。

先根据数据源key的分布情况或者分区分布情况，针对性的做一次repartition操作，重新存储，后续所有用到该数据源的程序都不会有数据倾斜的问题。

但是重分区预处理过程中仍然会存在数据倾斜问题而导致预处理过程缓慢。

如果该数据源只有当前程序使用，那么这个重分区预处理的操作就相当于在读取数据源的时候调用了repartition重分区、或者使用类似reduceByKey(500)调整并行度，实际上并没有起到多少作用。

所以，重分区预处理的方式只有在 一个数据源被n多个程序使用的时候比较有价值，使用的程序越多性价比越高，否则就是治标不治本，效果有限。

3.3.3 预聚合

如果前面两种方式都无法解决你的问题，而且 产生数据倾斜的Shuffle操作符是聚合类的（group、reduce、distinct）等，那么你可以尝试使用 预聚合 的方式。

还记得Mapreduce的Combiner吗，还记得reduceByKey和groupByKey的区别吗，不记得的话建议浏览一下「开发调优」中算子优化技巧。

Mapreduce的Combiner和Spark中的reduceByKey都会在各个节点的本地做一次预聚合。

类似的，如果存在某个key占据了绝大部分的数据量的话，我们也可以 手动采用预聚合的方式来分散热点数据并执行本地预聚合。

假设我们现在有1000w的数据，其中800w的数据都是相同的key，此时我们要做聚合操作，默认情况下800w数据会到同一个Task中处理，这肯定是无法接受的。

怎么手动做预聚合呢？

首先我们可以在这800w的key之前 根据任意hash算法添加固定长度的随机前缀。

在第一轮聚合时，这个热点key将会被打散到各个节点上去计算。

之后将key上的固定长度前缀去除，执行第二轮聚合操作。

因为经过第一轮聚合之后 热点key的数据已经被处理很多了，所以在第二轮聚合的时候可以比较轻松的处理。

当然如果在第二轮聚合的时候仍然有很大的热点问题，那么理论上可以 继续无限做预聚合处理。

但是预聚合的缺陷也很明显，只能优化聚合类的操作，如果是join等关联类的Shuffle操作则无法优化。

3.3.4 使用广播变量代替join

那么碰到join类的算子且发生了数据倾斜该如何处理呢？

其实我们在「开发调优」中已经提到过解决方式了，就是 使用广播变量来代替join操作。

但是这个方法也有很多限制，就是 只能应用于大表 join 小表的情况。

3.3.5 多种方案组合使用

如果以上的方案都没有能够解决你的问题，那么你可以尝试着将多种方案整合起来一起使用，因为在复杂的业务中，Shuffle操作符可能有很多，那么对应的可能产生数据倾斜的地方也有很多。

所以需要开发人员能够根据 业务逻辑、数据状态、代码编写 等方面能够根据不同的情况组合不同的方案来实施优化。

3.4 Shuffle调优

在上一节中我们着重介绍了如何针对「数据倾斜」这一情况进行优化。

除了数据倾斜可以优化之外，Shuffle过程中仍然有许多地方可以优化。但是要记住，影响Spark程序性能的主要因素还是在于 代码开发、资源参数与数据倾斜等，对Shuffle的调整优化可能仅仅是 锦上添花 而不是雪中送炭。

所以开发人员的重点应该放在前面几个部分，都优化完了之后可以考虑对Shuffle过程进行优化。

对Shuffle的优化主要是通过调整一些Shuffle相关的参数来实现，你可以根据你的使用情况和经验对以下参数进行调整：

• spark.Shuffle.file.buffer

  • 默认值：32k
  • 参数说明：用于设置Shuffle write task的BufferedOutputStream的buffer缓冲大小。将数据写到磁盘文件之前，会先写入buffer缓冲中，待缓冲写满之后，才会溢写到磁盘。
  • 调优建议：如果作业可用的 内存资源较为充足 的话，可以 适当增加这个参数的大小，从而减少Shuffle write过程中溢写磁盘文件的次数，也就可以减少磁盘IO次数，进而提升性能。

• spark.reducer.maxSizeInFlight

  • 默认值：48m
  • 参数说明：用于设置Shuffle read task的buffer缓冲大小，而这个buffer缓冲决定了每次能够拉取多少数据。
  • 调优建议：如果作业可用的 内存资源较为充足 的话，可以 适当增加这个参数的大小，从而减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能。

• spark.Shuffle.io.maxRetries

  • 默认值：3
  • 参数说明：Shuffle read task从Shuffle write task所在节点拉取属于自己的数据时，如果因为网络异常导致拉取失败，是会自动进行重试的。该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功，就可能会导致作业执行失败。
  • 调优建议：对于那些包含了 特别耗时的Shuffle操作的作业，建议 增加重试最大次数（比如60次），以避免 由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败，主要提升大型任务的执行稳定性。

• spark.Shuffle.io.retryWait

  • 默认值：5s
  • 参数说明：具体解释同上，该参数代表了每次重试拉取数据的等待间隔，默认是5s。
  • 调优建议：建议 加大间隔时长（比如60s），以增加Shuffle操作的稳定性。

• spark.Shuffle.manager

  • 默认值：sort
  • 参数说明：该参数用于设置ShuffleManager的类型。
  • 调优建议：由「Shuffle过程与管理」中可以知道，SortShuffleManager默认会对数据进行排序，如果程序中需要排序，那么使用默认即可；如果程序中不需要排序，那么建议 增大bypass的阈值以触发bypass机制或者将manager调整为hash，避免排序带来的开销，同时提供较好的磁盘读写性能。

• spark.shuffle.sort.bypassMergeThreshold

  • 默认值：200
  • 参数说明：当manager为sort，且Shuffle read task的数量小于这个阈值时，将会使用bypass机制。
  • 调优建议：使用sort manager时，如果不需要排序，那么就适当增加这个值，大于Shuffle read task的数量。

• spark.shuffle.consolidateFiles

  • 默认值：false
  • 参数说明：如果使用hash manager，该参数有效。如果设置为true，那么就会开启 consolidate机制，可以极大地减少磁盘IO开销，提升性能。
  • 调优建议：在不需要排序的情况下，除了使用sort manager触发bypass机制外，使用 hash manager + consolidate机制也是一个高性能的选择，建议使用此组合。

3.5 内存调优

和Shuffle调优一样，内存调优也是属于锦上添花的操作。

一般情况下，Spark能够自己很好的调整内存的使用，但是如果你对程序的执行性能有苛刻的要求的话，可以参考一下配置选项针对性的进行内存优化。

首先，建议使用新的内存管理模式，所以接下来的配置建议都是基于 统一内存管理 模式的：

• spark.memory.fraction

  • 如果 spill 次数发生较多（可通过Web界面观察到），可以适当调大该值
  • 这样 execution 和 storage 的总可用内存变大，能有效减少发生 spill 的频率

• spark.memory.storageFraction

  • 虽然新的内存管理方案中可以进行动态调整，但是提前根据应用情况（更多storage还是更多execution）设置内存占比也可以省略一些开销

• spark.memory.offHeap.enabled

  • 是否使用堆外内存
  • 堆外内存最大的好处就是可以避免 GC，需要将该值置为 true 并设置堆外内存的大小（spark.memory.offHeap.size）
  • 堆外内存的大小不会算在 executor memory 中，--executor memory 10G 和 spark.memory.offHeap.size=10G，那么总共可用20G

除了通过参数设置外，在编码过程中遵守以下几点也可以减少内存开销：

• 使用基本数据类型
• 减少包含大量小对象的嵌套结构
• key使用数值或者枚举而不是字符串

---

四、Spark最佳实践

4.1 编码技巧

4.1.1 Spark-DB-Connector

开始安利一下这个小组件了，使用Scala的简洁风格操作数据库，你可以在Spark程序中这样子读HBase：

val rdd = sc.fromHBase[(String, String, String)]("mytable") .select("col1", "col2") .inColumnFamily("columnFamily") .withStartRow("startRow") .withEndRow("endRow") //rdd type:RDD[(String,String,String)]

这样子写HBase：

//rdd type:RDD[(String,String,String)] rdd.toHBase("mytable") .insert("col1", "col2") .inColumnFamily("columnFamily") .save()

写MySQL的时候歪了，没有直接操作RDD，而是操作集合（貌似这样所有Scala程序都能用它读写MySQL）

读MySQL：

//以Spark中为例 val list = sc.fromMysql[(Int,String,Int)]("table-name") .select("id","name","age") .where("where-conditions") .get //list type:Seq[(Int,String,Int)]

写MySQL：

//list type:Iterable[T] list.toMysql("table-name") .insert("columns") .where("where-conditions") .save()

源代码在此：chubbyjiang/Spark_DB_Connector

4.1.2 关于filter和union

关于filter和union
这俩货还是不要凑一起的好...

如果你这样子操作：

val data1 = rdd.filter(condition1) val data2 = rdd.filter(condition2) ... val dataN = rdd.filter(conditionN) val totalData = data1.union(data2)...union(dataN).otherTransformation

去WebUI瞧一瞧吧，有个大大的surprise。

（多个filted RDD 进行union操作，产生的新RDD将会有n多个分区，而且大部分分区是无效的。）

4.1.3 将广播变量类型作为类成员属性

这块有个坑，假设在一个class中声明一个成员变量的类型为Broadcast[T]，然后在外部通过obj.bro = sc.broadcast(something)来设置的话，引用时为空。

broadcast操作会将要广播的数据存在某个地方（磁盘&&内存），并且开启一个类似HttpServer的服务允许各个节点上的Executor来Driver端获取数据，当task在各个节点上反序列化开始执行并使用到了广播变量之后就会到Driver那把数据拿到本地使用。

当使用obj.bro的时候是一个指向某个内存地址的指针，猜想跟上面的过程不搭嘎，so，无法使用。

4.1.4 foreachPartition

号外号外，如果 (foreachPartition操作的数据很大 or 单Executor内存不太够 ) and 你这样子做的话：

rdd.foreachPartition{ x.toList.//在list上面的其他操作 }

会惊喜的得到一份OOM（或者任务失败）大礼包，因为 toList操作把当前分区中的所有数据一次性加载到内存中处理。

主要临床症状表现为：

• 一开始的几个分区可以顺利完成操作
• 随着数据越读越多，Executor中的内存开始吃力的时候触发GC
• WebUI上各个Executor的GC时间特别长
• task停顿导致无法响应心跳检测从而任务失败

（回想一下之前的「Shuffle调优」和「内存调优」，如果要强行执行的话可以设置哪些参数？）

4.2 应用执行技巧

4.2.1 并行度和任务数量调整

在Yarn管理模式下一些启动的参数可以决定程序占多少资源。

集群启动的Container数量是根据资源分配的，如果设置的参数太高 导致资源不足 就只能起 最大限制的Container，此时有可能使得资源利用不高。

例如，集群有24个核心，48g的内存，启动Executor数量为24，每个Executor内存为2g，此时因为内存限制的原因，集群只会15个Container，导致实际上只用到了15个cpu。

如果想达到最大CPU/内存利用率，可以多测试几下，例如上面的集群中20个Executor，内存为1650m可以启动20个Container，内存使用41g，cpu利用率也达到了21，包括Driver进程。

当然，也不能一个程序就把所有资源吃掉，可以以这个最大限度作参考，好歹知个底。

可以简单的这样估算集群最大承受能力：

• 根据集群总CPU核心数估算启动多少个Executor，最好保留几个核心
• 使用总内存（减去Driver内存）除以得到的Executor数量估算每个Executor可以使用的内存，一般来说取这个内存的2/3，因为有其他保留内存限制，超出之后可能只会启动少数Container（可以多测试几下看最大启动的内存限制是多少）
• 根据Executor的情况，合理设置作业的task任务数

4.2.2 WebUI

这是个好东西啊，你想要的和不想要的都在上面了。

• 如果看到各个Executor的Input数据不平衡

  • 你可能有数据倾斜的麻烦

• 有特定的Stage中的Input比较不均匀

  • 会不会有filter这种会造成局部数据大小偏差的操作

• 有些Executor的GC时间长

  • 是不是很多数据往内存里面挤或者其他数据问题

UI中还可以看到任务执行的 各个阶段所以及其占据的总体时间，可以从中看出有问题的阶段从而进行排查调优。

4.2.3 CLI使用

Spark提供的CLI环境（Scala/Python/SQL）是一个十分强大的工具，在这个环境里面，执行的代码 所见即所得。

在CLI中你可以尽情的调试代码并查看输出的数据，在 程序调试、错误排查、数据分析 等场景中十分方便。

以Scala版本为例，为了方便将代码贴到CLI中执行并查看结果，各位同学在IDE中开发代码的时候可以注意一下以下几点：

1、长代码段换行时不要根据.来换行

如下：

//错误换行，CLI中将会识别成2个代码段，导致直接黏贴过去会执行失败 rdd.map .filter(x=> ) //正确换行，可以直接无脑复制黏贴执行 rdd.map.filter( x=> )

2、函数体或者类代码段之间的空行不超过2行

如下：

//错误的函数体，代码段之间空行超过2，CLI中将会识别成2个代码段，导致直接黏贴过去会执行失败 def test{ val a = 1 val b = 2 } //正确的函数体，代码段之间空行不超过2，可以直接无脑复制黏贴执行 def test{ val a = 1 val b = 2 }

3、待补充。。。

上面提到的技巧仅仅是为了能够将代码直接黏贴到CLI中无脑执行，如果不符合你们公司的代码规范，或者你不适用CLI的话可以无视（当然墙裂建议使用CLI）。

4.3 数据处理技巧

4.3.1 大量小文件的处理

小文件一多，直接读起来处理的话生成的task数量会让人目瞪口呆 ，处理的方式也不复杂，能在文件生成的时候就规定好大小是最直接的方法。

如果产生文件的系统不受你的控制那就考虑一下要不要把它们读起来合并一份。

如果程序开始处理前不能很好的搞定的话，那就在代码上动动手脚，RDD执行操作前加个coalesce(numPartitions,false)吧，不Shuffle的情况下也可以有效的减低task数量，不过数据可能不会均匀的分布在设置的partition中，因为不Shuffle的话就只是处理local的数据了。

4.3.2 RDD缓存

大数据量的RDD不合适持久化，有可能操作任务失败拖慢进度。

尽量不要直接使用原始数据集进行缓存，取出需要的内容减少体积之后再缓存。

哦，还有一个，最好不要缓存对象。

最后，记得 用一个变量保存rdd.cache的返回值，不然你怎么action都没用。

4.3.3 Shuffle相关

涉及到网络的一般都是程序中的瓶颈（毕竟和远程取数据相比，本地数据都可以看成是高速缓存了）。

少年，下手的时候留意一下你屏幕中的Shuffle操作。

稍微留意一下可以让Shuffle如丝般顺滑：

• spark.shuffle.consolidateFiles
• spark.shuffle.manager
• 尽量减少Shuffle操作：其实很多Shuffle操作符都可以通过其他算子来替换的，比如groupByKey->reduceByKey/aggregateByKey，或者其他组合拳
• 不能避免的情况下：尽量减少Shuffle前的数据集，没用的字段都扔了吧，能不用对象还是不要用了
• 参考「Shuffle优化」小节