spark性能调优（二） 完全解密spark的Hash Shuffle

装载：http://www.cnblogs.com/jcchoiling/p/6431969.html

引言

Spark HashShuffle 是它之前的版本，如今1.6x 版本默应是 Sort-Based Shuffle，那为何要讲 HashShuffle 呢，由于有分布式就必定会有 Shuffle，并且 HashShuffle 是 Spark之前的版本，亦便是 Sort-Based Shuffle 的前身，由于有 HashShuffle 的不足，才会有后续的 Sorted-Based Shuffle，以及如今的 Tungsten-Sort Shuffle，因此咱们有必要去了解它。

人们对Spark的印象每每是基于内存进行计算，但实际上来说，Spark能够基于内存、也能够基于磁盘或者是第三方的储存空间进行计算，背后有两层含意，第1、Spark框架的架构设计和设计模式上是倾向于在内存中计算数据的，第2、这也表达了人们对数据处理的一种美好的愿望，就是但愿计算数据的时候，数据就在内存中。

为何再一次强调 Shuffle 是 Spark 的性能杀手啦，那不就是说，Spark中的 “Shuffle“ 和 “Spark彻底是基于内存计算“ 的愿景是相违背的！！！但愿这篇文章能为读者带出如下的启发：

• 了解为何 Shuffle 是分布式系统的天敌
• 了解 Spark HashShuffle的原理和机制
• 了解优化后 Spark Consolidated HashShuffle的原理和机制
• 了解Shuffle 是如何成为 Spark 性能杀手
• 了解能够从那几方面思考 Spark Shuffle 的性能调优
• 了解 Spark HashShuffle 在读、写磁盘这个过程的源码鉴赏

 

1、shuffle是分布式系统的天敌

Spark 运行分红两部份，第一部份是 Driver Program，里面的核心是 SparkContext，它驱动著一个程序的开始，负责指挥，另一部份是 Worker 节点上的 Task，它是实际运行任务的，当程序运行时，不间断地由 Driver 与所在的进程进行交互，交互什么，有几点，第1、是让你去干什么，第2、是具体告诉 Task 数据在那里，例如说有三个 Stage，第二个 Task 要拿数据，它就会向 Driver 要数据，因此在整个工做的过程当中，Executor 中的 Task 会不断地与 Driver 进行沟通，这是一个网络传输的过程。

在这个过程当中一方面是 Driver 跟 Executor 进行网络传输，另外一方面是Task要从 Driver 抓取其余上游的 Task 的数据结果，因此有这个过程当中就不断的产生网络结果。其中，下一个 Stage 向上一个 Stage 要数据这个过程，咱们就称之为 Shuffle。

思考点：上一个 Stage 为何要向下一个 Stage 发数据？假设如今有一个程序，里面有五个 Stage，我把它当作为一个很大的 Stage，在分布式系统中，数据分布在不一样的节点上，每个节点计算一部份数据，若是不对各个节点上独立的部份进行汇聚的话，咱们是计算不到最终的结果。这就是由于咱们须要利用分布式来发挥它自己并行计算的能力，然后续又须要计算各节点上最终的结果，因此须要把数据汇汇集中，这就会致使 Shuffle，这也是说为何 Shuffle 是分布式不可避免的命运。

 

2、spark中的Hash Shuffle介绍

      一、  原始的Hash Shuffle机制

基于 Mapper 和 Reducer 理解的基础上，当 Reducer 去抓取数据时，它的 Key 究竟是怎么分配的，核心思考点是：做为上游数据是怎么去分配给下游数据的。在这张图中你能够看到有4个 Task 在2个 Executors 上面，它们是并行运行的，Hash 自己有一套 Hash算法，能够把数据的 Key 进行从新分类，每一个 Task 对数据进行分类而后把它们不一样类别的数据先写到本地磁盘，而后再通过网络传输 Shuffle，把数据传到下一个 Stage 进行汇聚。

 

下图有3个 Reducer，从 Task 开始那边各自把本身进行 Hash 计算，分类出3个不一样的类别，每一个 Task 都分红3种类别的数据，刚刚提过由于分布式的关系，咱们想把不一样的数据汇聚而后计算出最终的结果，因此下游的 Reducer 会在每一个 Task 中把属于本身类别的数据收集过来，汇聚成一个同类别的大集合，抓过来的时候会首先放在内存中，但内存可能放不下，也有可能放在本地 (这也是一个调优势。能够参考上一章讲过的一些调优参数)，每1个 Task 输出3份本地文件，这里有4个 Mapper Tasks，因此总共输出了4个 Tasks x 3个分类文件 = 12个本地小文件。

Hash Shuffle 也有它的弱点：

1. Shuffle前在磁盘上会产生海量的小文件，此时会产生大量耗时低效的 IO 操做 (因為产生过多的小文件）
2. 内存不够用，因为内存中须要保存海量文件操做句柄和临时信息，若是数据处理的规模比较庞大的话，内存不可承受，会出现 OOM 等问题。

   二、 优化后的Hash Shuffle 机制

    在刚才 HashShuffle 的基础上思考该如何进行优化，这是优化后的实现：

这里仍是有4个Tasks，数据类别仍是分红3种类型，由于Hash算法会根据你的 Key 进行分类，在同一个进程中，不管是有多少过Task，都会把一样的Key放在同一个Buffer里，而后把Buffer中的数据写入以Core数量为单位的本地文件中，(一个Core只有一种类型的Key的数据)，每1个Task所在的进程中，分别写入共同进程中的3份本地文件，这里有4个Mapper Tasks，因此总共输出是 2个Cores x 3个分类文件 = 6个本地小文件。Consoldiated Hash-Shuffle的优化有一个很大的好处就是假设如今有200个Mapper Tasks在同一个进程中，也只会产生3个本地小文件； 若是用原始的 Hash-Based Shuffle 的话，200个Mapper Tasks 会各自产生3个本地小文件，在一个进程已经产生了600个本地小文件。3个对比600已是一个很大的差别了。

这个优化后的 HashShuffle 叫 ConsolidatedShuffle，在实际生产环境下能够调如下参数：

spark.shuffle.consolidateFiles=true

　Consolidated HashShuffle 也有它的弱点： 

1. 若是 Reducer 端的并行任务或者是数据分片过多的话则 Core * Reducer Task 依旧过大，也会产生不少小文件。

     三、Shuffle是如何成为Spark性能杀手及调优势思考

Shuffle 不能够避免是由于在分布式系统中的基本点就是把一个很大的的任务/做业分红一百份或者是一千份，这一百份和一千份文件在不一样的机器上独自完成各自不一样的部份，咱们是针对整个做业要结果，因此在后面会进行汇聚，这个汇聚的过程的前一阶段到后一阶段以致网络传输的过程就叫 Shuffle。在 Spark 中为了完成 Shuffle 的过程会把真正的一个做业划分为不一样的 Stage，这个Stage 的划分是跟据依赖关系去决定的，Shuffle 是整个 Spark 中最消耗性能的一个地方。试试想一想若是没有 Shuffle 的话，Spark能够完成一个纯内存式的操做。

reduceByKey，它会把每一个 Key 对应的 Value 聚合成一个 value 而后生成新的 RDD

　　

Shuffle 是如何破坏了纯内存操做呢，由于在不一样节点上咱们要进行数据传输，数据在经过网络发送以前，要先存储在内存中，内存达到必定的程度，它会写到本地磁盘，(在之前 Spark 的版本它没有Buffer 的限制，会不断地写入 Buffer 而后等内存满了就写入本地，如今的版本对 Buffer 多少设定了限制，以防止出现 OOM，减小了 IO)

Mapper 端会写入内存 Buffer，这个便关乎到 GC 的问题，而后 Mapper端的 Block 要写入本地，大量的磁盘与IO的操做和磁盘与网络IO的操做，这就构成了分布式的性能杀手。

若是要对最终计算结果进行排序的话，通常会都会进行 sortByKey，若是以最终结果来思考的话，你能够认为是产生了一个很大很大的 partition，你能够用 reduceByKey 的时候指定它的并行度，例如你把 reduceByKey 的并行度变成为1，新 RDD 的数据切片就变成1，排序通常都会在不少节点上，若是你把不少节点变成一个节点而后进行排序，有时候会取得更好的效果，由于数据就在一个节点上，技术层面来说就只须要在一个进程里进行排序。

能够在调用 reduceByKey()接著调用 mapPartition( )；
也能够用 repartitionAndSortWithPartitions( )； 

　　还有一个很危险的地方就是数据倾斜，在咱们谈的 Shuffle 机制中，不断强调不一样机器从Mapper端抓取数据并计算结果，但有没有意会到数据可能会分布不均衡，何时会致使数据倾斜，答案就是 Shuffle 时会导政数据分布不均衡，也就是数据倾斜的问题。数据倾斜的问题会引伸不少其余问题，好比，网络带宽、各重硬件故障、内存过分消耗、文件掉失。由于 Shuffle 的过程当中会产生大量的磁盘 IO、网络 IO、以及压缩、解压缩、序列化和反序列化等等。

    四、Shuffle 性能调优思考

Shuffle可能面临的问题，运行 Task 的时候才会产生 Shuffle (Shuffle 已经融化在 Spark 的算子中)

1. 几千台或者是上万台的机器进行汇聚计算，数据量会很是大，网络传输会很大
2. 数据如何分类其实就是 partition，即如何 Partition、Hash 、Sort 、计算
3. 负载均衡 (数据倾斜）
4. 网络传输效率，须要压缩或解压缩之间作出权衡，序列化 和 反序列化也是要考虑的问题

具体的 Task 进行计算的时候尽一切最大可能使得数据具有 Process Locality 的特性，退而求其次是增长数据分片，减小每一个 Task 处理的数据量，基于Shuffle 和数据倾斜所致使的一系列问题，能够延伸出不少不一样的调优势，好比说：

• Mapper端的 Buffer 应该设置为多大呢？
• Reducer端的 Buffer 应该设置为多大呢？若是 Reducer 太少的话，这会限制了抓取多少数据
• 在数据传输的过程当中是否有压缩以及该用什么方式去压缩，默应是用 snappy 的压缩方式。
• 网络传输失败重试的次数，每次重试之间间隔多少时间。