1.分配更多的资源 -- 性能调优的王道

真实项目里的脚本：

    bin/spark-submit \

    --class com.xx.xx \

    --num-executors 80 \

    --driver-memory 6g \

    --executor-cores 3 \

    --master yarn-cluster \

    --queue root.default \

    --conf spark.yarn.executor.memoryOverhead=2048 \

    --conf spark.core.connection.ack.waite.timeout=300 \

    /usr/xx/xx.jar \

    args

分配资源之前，要先看机器能够给我们提供多少资源，并且最大化的利用这些资源才是最好的；

1.standalone模式：

    根据实际情况分配spark作业资源，比如每台机器4G，2个cpu，20台机器

2.spark-on-yarn模式：

    要看spark作业要提交到的资源队列，大概有多少资源？

SparkContext、DAGScheduler、taskScheduler，会将我们的算子切割成大量的task提交到Application的executor上去执行,比如分配给其中一个的executor100个task，他的cpu只有2个，那么并行只能执行2个task，如果cpu为5个，那么久先执行5个再执行5个

a.增加executor的数量：

    如果executor的数量比较少，那么意味着可以并行执行的task的数量就比较少，

    就意味着Application的并行执行能力比较弱；

        比如：

            有3个executor，每个executor有2个cup core，

            那么能够并行执行的task就是6个，6个执行完换下一批6个

    增加executor的个数后，并行执行的task就会变多，性能得到提升

b.增加每个executor的cpu core

        比如：

            之前：3个exexutor，每个executor的cpu core为2个，那么并执行的task是6个

                把cpu core增加到5个，那么并行执行的task就是15个，提高了性能

c.增加每个executor的内存量：

    1.如果需要对RDD进行cache，那么更多的内存就能缓存更多的数据，

      将更少的数据写入磁盘，甚至不写入磁盘，减少了磁盘IO。

    2.对于shuffle操作，reduce端会需要内存来存储拉取过来的数据进行聚合，如果内存不够，

      也会写入磁盘，增加executor内存，就会有更少的数据写入磁盘，较少磁盘IO，提高性能。

    3.对于task的执行，需要创建很多对象，如果内存比较小，可能导致JVM堆内存满了，

      然后频繁的GC，垃圾回收，minorGC和fullGC，速度会很慢，如果加大内存，

      带来更少的GC，速度提升。

2.调节并行度

并行度：spark作业中，各个stage的task个数，也就代表了saprk作业中各个阶段[stage]的并行度。

[spark作业，Application，jobs，action会触发一个job，每个job会拆成多个stage，发生shuffle的时候，会拆出一个stage]

如果不调节并行度，导致并行度过低，会怎么样？？？

比如：

    1.我们通过上面的分配好资源后，有50个executor，每个executor10G内存，每个executor有3个cpu core

      基本已经达到了集群或者yarn的资源上限

    2. 50个executor * 3个cpu = 150个task,即可以并行执行150个task；

      而我们设置的时候只设置了100个并行度的task，这时候每个executor分配到2个task，同时运行task的数量只有100个，导致每个executor剩下的1个cpu core在那空转，浪费资源。

    资源虽然够了，但是并行度没有和资源相匹配，导致分配下去的资源浪费掉了！！！

    **合理的并行度设置，应该要设置的足够大，大到完全合理的利用集群资源！而且减少了每个task要处理的数据量[比如150g的数据分别分发给100个task处理就是每个task处理1.5G，但是如果是150个task的话，每个task就处理1G]**

总结：

    a. task数量，至少设置成与Spark application的总cpu数相同(理想状态下，比如150个cpu core，分配150个task，差不多同时运行完毕)

    b.官方推荐做法：task的数量设置成 Spark application的cpu core的个数的3~5倍！！

      比如总共150个cpu core，设置成300~500个task

      为什么这么设置呢？？？

        实际情况下和理想状态下是不一样的，因为有些task运行的快，有些运行的慢，

        比如有些运行需要50s，有些需要几分钟运行完毕，如果刚好设置task数量和cpu core的数量相同，可能会导致资源的浪费；

        比如150个task，10个运行完了，还有140个在运行，那么势必会导致10个cpu core的闲置，

        所以如果设置task的数量为cpu数量的2~3倍，一旦有task运行完，另一个task就会立刻补上来，

        尽量让cpu core不要空闲，提升了spark作业运行效率，提升性能。

    c.如何设置一个 Spark application的并行度？？？

        SparkConf sc = new SparkConf()

                      .set("spark.default.parallelism","500");

3.重构RDD架构及RDD持久化:

默认情况下 RDD出现的问题：           

                                              RDD4

                                            /

          hdfs --> RDD1 --> RDD2 -->RDD3

                                            \

                                              RDD5

    以上情况： 执行RDD4和RDD5的时候都会从RDD1到RDD2然后到RDD3，执行期间的算子操作，

    而不会说到RDD3的算子操作后的结果给缓存起来，所以这样很麻烦，

    出现了RDD重复计算的情况，导致性能急剧下降！

结论：

    对于要多次计算和使用的公共RDD,一定要进行持久化！

    持久化也就是：BlockManager将RDD的数据缓存到内存或者磁盘上，后续无论对这个RDD进行多少次计算，都只需要到缓存中取就ok了。

    持久化策略：

        rdd.persist(StorageLevel.xx()) 或者 cache

        1.优先会把数据缓存到内存中 -- StorageLevel.MEMORY_ONLY

        2.如果纯内存空间无法支撑公共RDD的数据时，就会优先考虑使用序列化的机制在纯内存中存储，

        将RDD中的每个partition中的数据序列化成一个大的字节数组，也就是一个对象，

        序列化后，大大减少了内存空间的占用。-- StorageLevel.MEMORY_ONLY_SER

            序列化唯一的缺点：在获取数据的时候需要反序列化

        3.如果序列化纯内存的方式还是导致OOM，内存溢出的话，那就要考虑磁盘的方式。

          内存+磁盘的普通方式(无序列化) -- StorageLevel.MEMORY_AND_DISK

        4.如果上面的还是无法存下的话，就用 内存+磁盘+序列化 -- StorageLevel.MEMORY_AND_DISK_SER

另：在机器内存**极度充足**的情况下，可以使用双副本机制，来持久化，保证数据的高可靠性

    如果机器宕机了，那么还有一份副本数据，就不用再次进行算子计算了。[锦上添花--一般不要这么做] -- StorageLevel.MEMORY_ONLY_SER_2

4.广播大变量 [sc.broadcast(rdd.collect)]

问题情景：

    当我们在写程序用到外部的维度表的数据进行使用的时候，程序默认会给每个task都发送相同的这个数据，

    如果这个数据为100M，如果我们有1000个task，100G的数据，通过网络传输到task，

    集群瞬间因为这个原因消耗掉100G的内存，对spark作业运行速度造成极大的影响，性能想想都很可怕！！！

解决方案：

    sc.broadcast(rdd.collect)

    分析原理：

        [BlockManager:负责管理某个executor上的内存和磁盘上的数据]

        广播变量会在Driver上有一份副本，当一个task使用到广播变量的数据时，会在自己本地的executor的BlockManager去取数据，

        发现没有，BlockManager就会到Driver上远程去取数据，并保存在本地，

        然后第二个task需要的时候来找BlockManager直接就可以找到该数据，

        executor的Blockmanager除了可以到Driver远程的取数据，

        还可能到邻近节点的BlockManager上去拉取数据，越近越好!

    举例说明：

        50个executor，1000个task，外部数据10M，

        默认情况下，1000个task，1000个副本，10G数据，网络传输，集群中10G的内存消耗

        如果使用广播，50个executor，500M的数据，网络传输速率大大增加，

        10G=10000M 和 500M的对比 20倍啊。。。

**之前的一个测试[真实]：

        没有经过任何调优的spark作业，运行下来16个小时，

        合理分配资源+合理分配并行度+RDD持久化，作业下来5个小时,

        非常重要的一个调优Shuffle优化之后，2~3个小时，

        应用了其他的性能调优之后，JVM调参+广播等等，30分钟

        16个小时 和 30分钟对比，太可怕了！！！性能调优真的真的很重要！！！

5.Kryo序列化机制：

默认情况下，Spark内部使用java的序列化机制

ObjectOutPutStream/ObjectInputStream,对象输入输出流机制来进行序列化

这种序列化机制的好处：

    处理方便，只是需要在算子里使用的变量是实现Serializable接口即可

缺点在于：

    默认序列化机制效率不高，序列化的速度比较慢，序列化后的数据占用内存空间还比较大

解决：手动进行序列化格式的优化：Kryo [spark支持的]

Kryo序列化机制比默认的java序列化机制速度快，

序列化后的数据更小，是java序列化后数据的 1/10 。

所以Kryo序列化后，会让网络传输的数据更少，在集群中耗费的资源大大减少。

Kryo序列化机制：[一旦启用，会生效的几个地方]

    a.算子函数中使用到的外部变量[比如广播的外部维度表数据]

        优化网络传输性能，较少集群的内存占用和消耗

    b.持久化RDD时进行序列化，StorageLevel.MEMORY_ONLY_SER

        将每个RDD partition序列化成一个大的字节数组时，就会使用Kryo进一步优化序列化的效率和性能。

        持久化RDD占用的内存越少，task执行的时候，创建的对象，不至于频繁的占满内存，频繁的GC

    c.shuffle

        在stage间的task的shuffle操作时，节点与节点之间的task会通过网络拉取和传输数据，

        此时这些数据也是可能需要序列化的，就会使用Kryo

实现Kryo：

step1. 在SparkConf里设置 new SparkConf()

                      .set("spark.serializer","org.apache.spark.serializer.KyroSerializer")

                      .registerKryoClasses(new Class[]{MyCategory.class})

    [Kryo之所以没有没有作为默认的序列化类库，就是因为Kryo要求，如果要达到它的最佳效果的话]

    [一定要注册我们自定义的类，不如：算子函数中使用到了外部自定义的对象变量，这时要求必须注册这个类，否则Kyro就达不到最佳性能]

step2. 注册使用到的，需要Kryo序列化的一些自定义类

6.使用FastUtil优化数据格式：

FastUtil是什么？？

fastutil是扩展了Java标准集合框架（Map、List、Set；HashMap、ArrayList、HashSet）的类库，提供了特殊类型的map、set、list和queue；

fastutil能够提供更小的内存占用，更快的存取速度；我们使用fastutil提供的集合类，来替代自己平时使用的JDK的原生的Map、List、Set，

fastutil的每一种集合类型，都实现了对应的Java中的标准接口（比如fastutil的map，实现了Java的Map接口），因此可以直接放入已有系统的任何代码中。

fastutil还提供了一些JDK标准类库中没有的额外功能（比如双向迭代器）。

fastutil除了对象和原始类型为元素的集合，fastutil也提供引用类型的支持，但是对引用类型是使用等于号（=）进行比较的，而不是equals()方法。

fastutil尽量提供了在任何场景下都是速度最快的集合类库。

Spark中FastUtil运用的场景？？

1.如果算子函数使用了外部变量，

    第一可以使用broadcast广播变量优化；

    第二可以使用Kryo序列化类库，提升序列化性能和效率；

    第三如果外部变量是某种比较大的集合(Map、List等)，可以考虑fastutil来改写外部变量，

        首先从源头上就减少了内存的占用，通过广播变量进一步减少内存占用，

        再通过Kryo类库进一步减少内存占用

    避免了executor内存频繁占满，频繁唤起GC，导致性能下降的现象

使用步骤：

step1:导入pom依赖

    <dependency>

        <groupId>fastutil</groupId>

            <artifactId>fastutil</artifactId>

        <version>5.0.9</version>

    </dependency>

step2:

    List<Integer> => IntList

    基本都是类似于IntList的格式，前缀就是集合的元素类型，

    特殊的就是Map，Int2IntMap，代表了Key-Value映射的元素类型

7.调节数据本地化等待时长:

问题发生的场景：

spark在Driver上，对Application的每一个stage的task分配之前，

都会计算出每个task要计算的是哪个分片数据，RDD的某个partition；

spark的分配算法：

    a.优先把每一个task正好分配到他要计算的数据所在的节点，这样的话不用在网络间传输数据

    b.但是，task没有机会分配到数据所在的节点上，为什么呢？？？

        因为那个节点上的计算资源和计算能力都满了，这个时候 spark会等待一段时间，

        默认情况下是3s钟，到最后，实在等不了了，就会选择一个较差的本地化级别，

        比如说会把task分配到靠他要计算的数据的节点最近的节点，然后进行计算

    c.对于b来说肯定要发生网络传输，task会通过其所在节点的executor的BlockManager来获取数据，

    BlockManager发现自己本地没有，就会用getRemote()的方法，通过TransferService(网络数据传输组件)

    从数据所在节点的BlockManager中获取数据，通过网络传输给task所在的节点

总结：

  我们肯定是希望看到 task和数据都在同一个节点上，直接从本地的executor中的BlockManager中去获取数据，

  纯内存或者带点IO，如果通过网络传输，那么大量的网络传输和磁盘IO都是性能的杀手

本地化的级别类型：

1.PROCESS_LOCAL: 进程本地化,代码和数据都在同一个进程中，也就是在同一个executor进程中，

  task由executor来执行，数据在executor的BlockManager中，性能最好

2.NODE_LOCAL: 节点本地化，比如说一个节点上有两个executor，其中一个task需要第一个executor的数据，

  但是他被分配给了第二个executor，他会找第二个executor的BlockManager去取数据，但是没有，

  BlockManager会去第一个的executor的BlockManager去取数据，这是发生在进程中的

3.NOPREF: 数据从哪里获取都一样，没有好坏之分

4.RACK_LOCAL: 数据在同一个机架上的不同节点上，需要进行网络间的数据传输

5.ANY: 数据可能在集群中的任何地方，而且不在同一个机架，这种性能最差！！

开发时的流程：

观察spark作业时的日志，先测试，先用client模式，在本地就可以看到比较全的日志。

日志里面会显示：starting task...,PROCESS_LOCAL,或者是NODE_LOCAL，观察大部分数据本地化的级别

如果发现大部分都是PROCESS_LOCAL的级别，那就不用调了，如果大部分都是NODE_LOCAL或者ANY，那就要调节一下等待时长了

要反复调节，反复观察本地化级别是否提升，查看spark作业运行的时间有没有缩短

不要本末倒置，如果是 本地化级别提升了，但是因为大量的等待时间，spark作业的运行时常变大了，这就不要调节了

spark.locality.waite

spark.locality.waite.process

spark.locality.waite.node

spark.locality.waite.rack

默认等待时长都是3s

设置方法：

    new SparkConf().set("spark.locality.waite","10")//不要带s

8.JVM调优：1个executor对应1个JVM进程

A. 降低cache操作的内存占比

JVM模块：

    每一次存放对象的时候都会放入eden区域，其中有一个survivor区域，另一个survivor区域是空闲的[新生代]，

    当eden区域和一个survivor区域放满了以后(spark运行产生的对象太多了)，

    就会触发minor gc，小型垃圾回收，把不再使用的对象从内存中清空，给后面新创建的对象腾出空间

    清理掉了不在使用的对象后，还有一部分存活的对象(还要继续使用的对象)，

    将存活的对象放入空闲的那个survivor区域里，这里默认eden:survivor1: survivor2 = 8:1:1,

    假如对象占了1.5放不下survivor区域了，那么就会放到[老年代]里；

    假如JVM的内存不够大的话，可能导致频繁的新生代内存满溢，频繁的进行minor gc，

    频繁的minor gc会导致短时间内，有些存活的对象，多次垃圾回收都没有回收掉，

    会导致这种短生命周期的对象(其实是不一定要长期使用的对象)年龄过大，

    垃圾回收次数太多，还没有回收到，就已经跑到了老年代；

    老年代中可能会因为内存不足，囤积一大堆短生命周期的对象(本来应该在年轻代中的)，

    可能马上就要回收掉的对象，此时可能造成老年代内存满溢，造成频繁的full gc(全局/全面垃圾回收)，full gc就会去老年代中回收对象；

    由于full gc算法的设计，是针对老年代中的对象，数量很少，满溢进行full gc的频率应该很少，

    因此采取了不太复杂的但是耗费性能和时间的垃圾回收算法。full gc 很慢很慢；

    full gc 和 minor gc，无论是快还是慢，都会导致JVM的工作线程停止工作，即 stop the world，

    简言之：gc的时候，spark停止工作，等待垃圾回收结束;

在spark中，堆内存被分为了两块：

    一块是专门用来给RDD cache和persist操作进行RDD数据缓存用的；

    一块是给spark算子函数的运行使用的，存放函数中自己创建的对象；

默认情况下，给RDD cache的内存占比是60%,但是在某些情况下，比如RDD cache不那么紧张，

而task算子函数中创建的对象过多，内存不太大，导致频繁的minor gc，甚至频繁的full gc，

导致spark频繁的暂停工作，性能影响会非常大，

解决办法：

    集群是spark-onyarn的话就可以通过spark ui来查看，spark的作业情况，

    可以看到每个stage的运行情况，包括每个task的运行时间，gc时间等等，

    如果发现gc太频繁，时间太长，此时可以适当调节这个比例；

总结：

    降低cache的内存占比，大不了用persist操作，选择将一部分的RDD数据存入磁盘，

    或者序列化方式Kryo，来减少RDD缓存的内存占比；

    对应的RDD算子函数的内存占比就增多了，就可以减少minor gc的频率，同时减少full gc的频率，提高性能

具体实现：0.6->0.5->0.4->0.2

    new SparkConf().set("spark.storage.memoryFraction","0.5")

B. executor堆外内存与连接时常

1. executor堆外内存[off-heap memory]:

  场景：

        比如两个stage，第二个stage的executor的task需要第一个executor的数据，

        虽然可以通过Driver的MapOutputTracker可以找到自己数据的地址[也就是第一个executor的BlockManager]，

        但是第一个executor已经挂掉了，关联的BlockManager也没了，就没办法获取到数据；

    有时候，如果你的spark作业处理的数据量特别大，几亿数据量；

    spark作业一运行，是不是报错诸如：shuffle file cannot find,executor task lost,out of memory,

    这时候可能是executor的堆外内存不够用了，导致executor在运行的时候出现了内存溢出；

    导致后续的stage的task在运行的时候，可能从一些executor中拉取shuffle map output 文件，

    但是executor已经挂掉了，关联的BlockManager也没有了，所以可能会报shuffle output file not found，resubmitting task，executor lost，spark作业彻底失败；

  这个时候就可以考虑调节executor的堆外内存，堆外内存调节的比较大的话，也会提升性能；

    怎么调价堆外内存的大小？？

        在spark-submit 的脚本中添加

                    --conf spark.yarn.executor.memoryOverhead=2048

        注意：这个设置是在spark-submit脚本中，不是在 new SparkConf()里设置的！！！

        这个是在spark-onyarn的集群中设置的，企业也是这么设置的！

        默认情况下,堆外内存是300多M，我们在项目中通常都会出现问题，导致spark作业反复崩溃，

        我们就可以调节这个参数 ，一般来说至少1G(1024M)，有时候也会2G、4G，

        来避免JVM oom的异常问题，提高整体spark作业的性能

2. 连接时常的等待：

        知识回顾：如果JVM处于垃圾回收过程，所有的工作线程将会停止，相当于一旦进行垃圾回收，

        spark/executor就会停止工作，无法提供响应

  场景：

        通常executor优先会从自己关联的BlockManager去取数据，如果本地没有，

        会通过TransferService，去远程连接其他节点上的executor的BlockManager去取；

        如果这个远程的executor正好创建的对象特别大，特别多，频繁的让JVM的内存满溢，进行垃圾回收，

        此时就没有反应，无法建立网络连接，会有卡住的现象。spark默认的网络连接超时时间是60s，

        如果卡住60秒都无法建立网络连接的话，就宣布失败；

        出现的现象：偶尔会出现，一串fileId诸如：hg3y4h5g4j5h5g5h3 not found，file lost，

        报错几次，几次都拉取不到数据的话，可能导致spark作业的崩溃！

        也可能会导致DAGScheduler多次提交stage，TaskScheduler反复提交多次task，

        大大延长了spark作业的运行时间

  解决办法：[注意是在shell脚本上不是在SparkConf上set！！]

        spark-submit

                    --conf spark.core.connection.ack.waite.timeout=300

9.shuffle调优

shuffle的概念以及场景

    什么情况下会发生shuffle？？

        在spark中，主要是这几个算子：groupByKey、reduceByKey、countByKey、join等

    什么是shuffle？

        a) groupByKey：把分布在集群中各个节点上的数据中同一个key，对应的values都集中到一块，

        集中到集群中的同一个节点上，更严密的说就是集中到一个节点上的一个executor的task中。

        集中一个key对应的values后才能交给我们处理，<key,iterable<value>>

        b) reduceByKey：算子函数对values集中进行reduce操作，最后变成一个value

        c) join  RDD<key,value>    RDD<key,value>,只要两个RDD中key相同的value都会到一个节点的executor的task中，供我们处理

    以reduceByKey为例：

9.1. shuffle调优之 map端合并输出文件

默认的shuffle对性能有什么影响？？

    实际生产环境的条件：

        100个节点，每个节点一个executor：100个executor，每个executor2个cpu core，

        总格1000个task，平均到每个executor是10个task；按照第二个stage的task个数和第一个stage的相同，

        那么每个节点map端输出的文件个数就是：10 * 1000 = 10000 个

        总共100个节点，总共map端输出的文件数：10000 * 100 = 100W 个

        100万个。。。太吓人了！！！

    shuffle中的写磁盘操作，基本上是shuffle中性能消耗最严重的部分，

    通过上面的分析可知，一个普通的生产环境的spark job的shuffle环节，会写入磁盘100万个文件，

    磁盘IO性能和对spark作业执行速度的影响，是极其惊人的！！

    基本上，spark作业的性能，都消耗在了shuffle中了，虽然不只是shuffle的map端输出文件这一部分，但是这也是非常大的一个性能消耗点。

怎么解决？

    开启map端输出文件合并机制：

        new SparkConf().set('spark.shuffle.consolidateFiles','true')

    实际开发中，开启了map端输出文件合并机制后，有什么变化？

        100个节点，100个executor，

        每个节点2个cpu core，

        总共1000个task，每个executor10个task，

        每个节点的输出文件个数：

            2*1000 = 2000 个文件

        总共输出文件个数：

            100 * 2000 = 20万 个文件

        相比开启合并之前的100万个，相差了5倍！！

合并map端输出文件，对spark的性能有哪些影响呢？

    1. map task写入磁盘文件的IO，减少：100万 -> 20万个文件

    2. 第二个stage，原本要拉取第一个stage的task数量文件，1000个task，第二个stage的每个task都会拉取1000份文件，走网络传输；合并以后，100个节点，每个节点2个cpu，第二个stage的每个task只需要拉取 100 * 2 = 200 个文件，网络传输的性能大大增强

    实际生产中，使用了spark.shuffle.consolidateFiles后，实际的调优效果：

        对于上述的生产环境的配置，性能的提升还是相当可观的，从之前的5个小时 降到了 2~3个小时

总结：

    不要小看这个map端输出文件合并机制，实际上在数据量比较大的情况下，本身做了前面的优化，

    executor上去了 -> cpu core 上去了 -> 并行度（task的数量）上去了，但是shuffle没调优，

    这时候就很糟糕了，大量的map端输出文件的产生，会对性能有比较恶劣的影响

9.2. map端内存缓冲与reduce端内存占比

spark.shuffle.file.buffer,默认32k

spark.shuffle.memoryFraction,占比默认0.2

调优的分量：

    map端内存缓冲和reduce端内存占比，网上对他俩说的是shuffle调优的不二之选，其实这是不对的，

    因为以实际的生产经验来说，这两个参数没那么重要，但是还是有一点效果的，

    就像是很多小的细节综合起来效果就很明显了，

原理：

map：

    默认情况下，shuffle的map task输出到磁盘文件的时候，统一都会先写入每个task自己关联的一个内存缓冲区中，

    这个缓冲区默认大小是32k，每一次，当内存缓冲区满溢后，才会进行spill操作，溢写到磁盘文件中

reduce：

    reduce端task，在拉取数据之后，会用hashmap的数据格式来对每个key对应的values进行汇聚，

    针对每个key对应的value，执行我们自定义的聚合函数的代码，比如_+_,(把所有values相加)

    reduce task,在进行汇聚、聚合等操作的时候，实际上，使用的就是自己对应的executor的内存，

    executor(jvm进程，堆),默认executor内存中划分给reduce task进行聚合的比例是20%。

    问题来了，内存占比是20%，所以很有可能会出现，拉取过来的数据很多，那么在内存中，

    放不下，这个时候就会发生spill(溢写)到磁盘文件中取.

如果不调优会出现什么问题？？

默认map端内存缓冲是32k，

默认reduce端聚合内存占比是20%

如果map端处理的数据比较大，而内存缓冲是固定的，会出现什么问题呢？

    每个task处理320k，32k的内存缓冲，总共向磁盘溢写10次，

    每个task处理32000k，32k的内存缓冲，总共向磁盘溢写1000次，

    这样就造成了多次的map端往磁盘文件的spill溢写操作，发生大量的磁盘IO，降低性能

map数据量比较大，reduce端拉取过来的数据很多，就会频繁的发生reduce端聚合内存不够用，

频繁发生spill操作，溢写到磁盘上去，这样一来，磁盘上溢写的数据量越大，

后面进行聚合操作的时候，很可能会多次读取磁盘中的数据进行聚合

默认情况下，在数据量比较大的时候，可能频繁的发生reduce端磁盘文件的读写；

这两点是很像的，而且有关联的，数据量变大，map端肯定出现问题，reduce也出现问题，

出的问题都是一样的，都是磁盘IO频繁，变多，影响性能

调优解决：

    我们要看spark UI，

        1. 如果公司用的是standalone模式，那么很简单，把spark跑起来，会显示sparkUI的地址，

        4040端口号，进去看，依次点击可以看到，每个stage的详情，有哪些executor，有哪些task，

        每个task的shuffle write 和 shuffle read的量，shuffle的磁盘和内存，读写的数据量

        2. 如果是yarn模式提交，从yarn的界面进去，点击对应的application，进入spark ui，查看详情

    如果发现磁盘的read和write很大，就意味着要调节一下shuffle的参数，进行调优，

    首先当然要考虑map端输出文件合并机制

    调节上面两个的参数，原则是：

        spark.shuffle.buffer，每次扩大一倍，然后看看效果，64k，128k

        spark.shuffle.memoryFraction,每次提高0.1，看看效果

    不能调节的过大，因为你这边调节的很大，相对应的其他的就会变得很小，其他环节就会出问题

    调节后的效果：

        map task内存缓冲变大了，减少了spill到磁盘文件的次数；

        reduce端聚合内存变大了，减少了spill到磁盘的次数，而且减少了后面聚合时读取磁盘的数量

        new SparkConf()

        .set("spark.shuffle.file.buffer","64")

        .set("spark.shuffle.file.memoryFraction","0.3")

10.算子调优

1.算子调优之MapPartitons提升map的操作性能

在spark中最近本的原则：每个task处理RDD中的每一个partition

优缺点对比：

    普通Map：

        优点：比如处理了一千条数据，内存不够了，那么就可以将已经处理的一千条数据从内存里面垃圾回收掉，

        或者用其他办法腾出空间；通常普通的map操作不会导致内存OOM异常；

        缺点：比如一个partition中有10000条数据，那么function会执行和计算一万次

    MapPartitions:

        优点：一个task仅仅会执行一次function，一次function接收partition中的所有数据

        只要执行一次就可以了，性能比较高

        缺点：对于大数据量来说，比如一个partition100万条数据，一次传入一个function后，

        可能一下子内存就不够了，但是又没办法腾出空间来，可能就OOM，内存溢出

那么什么时候使用MapPartitions呢？

    当数据量不太大的时候，都可以使用MapPartitions来操作，性能还是很不错的，

    不过也有经验表明用了MapPartitions后，内存直接溢出，

    所以在项目中自己先估算一下RDD的数据量，以每个partition的量，还有分配给executor的内存大小，

    可以试一下，如果直接OOM了，那就放弃吧，如果能够跑通，那就可以使用。

2.算子调优之filter之后 filter 之后 用 coalesce来减少partition的数量

默认情况下，RDD经过filter之后，RDD中每个partition的数据量会不太一样，(原本partition里的数据量可能是差不多的)

问题：

    1.每一个partition的数据量变少了，但是在后面进行处理的时候，

    还是要和partition的数量一样的task数量去处理，有点浪费task计算资源

    2.每个partition的数据量不一样，后面会导致每个处理partition的task要处理的数据量不一样，

    这时候很容易出现**数据倾斜**

    比如说，有一个partition的数据量是100，而另一个partition的数据量是900，

    在task处理逻辑一样的情况下，不同task要处理的数据量可能差别就到了9倍，甚至10倍以上，

    同样导致速度差别在9倍或者10倍以上

    这样就是导致了有的task运行的速度很快，有的运行的很慢，这就是数据倾斜。

解决：

    针对以上问题，我们希望把partition压缩，因为数据量变小了，partition完全可以对应的变少，

    比如原来4个partition，现在可以变成2个partition，那么就只要用后面的2个task来处理，

    不会造成task资源的浪费(不必要针对只有一点点数据的partition来启动一个task进行计算)

    避免了数据倾斜的问题

3.算子调优之使用foreachPartition优化写入数据库性能

默认的foreach有哪些缺点？

    首先和map一样，对于每条数据都要去调一次function，task为每个数据，都要去执行一次task；

    如果一个partition有100万条数据，就要调用100万次，性能极差！

    如果每条数据都要创建一个数据库连接，那么就要创建100万个数据库连接，

    但是数据库连接的创建和销毁都是非常耗性能的，虽然我们用了数据库连接池，只要创建固定数量的连接，

    还是得多次通过数据库连接，往数据库里(mysql)发送一条sql语句，mysql需要去执行这条sql语句，

    有100万条数据，那么就是要发送100万次sql语句；

用了foreachPartition以后，有哪些好处？

    1.对于我们写的函数就调用一次就行了，一次传入一个partition的所有数据

    2.主要创建或者获取一个数据库连接就可以了

    3.只要向数据里发送一条sql语句和一组参数就可以了

在实际开发中，我们都是清一色使用foreachPartition算子操作，

但是有个问题，跟mapPartitions操作一样，如果partition的数据量非常大，

比如真的是100万条，那几本就不行了！一下子进来可能会发生OOM，内存溢出的问题

一组数据的对比：

    生产环境中：

        一个partition中有1000条数据，用foreach，跟用foreachPartition，

        性能提高了2~3分；

数据库里是：

    for循环里preparestatement.addBatch

    外面是preparestatement.executeBatch

4.算子调优之repartition解决SparkSQL低并行度的问题

并行度： 我们是可以自己设置的

    1.spark.default.parallelism

    2.sc.textFile(),第二个参数传入指定的数量(这个方法用的非常少)

在生产环境中，我们是要自己手动设置一下并行度的，官网推荐就是在spark-submit脚本中，

指定你的application总共要启动多少个executor，100个，每个executor多少个cpu core，

2~3个，假设application的总cpu core有200个；

官方推荐设置并行度要是总共cpu core个数的2~3倍，一般最大值，所以是 600；

设置的这个并行度，在哪些情况下生效？哪些情况下不生效？

    1.如果没有使用SparkSQL(DataFrame)的话，那么整个spark应用的并行度就是我们设置的那个并行度

    2.如果第一个stage使用了SparkSQL从Hive表中查询了一些数据，然后做了一些transformatin的操作，

    接着做了一个shuffle操作(groupByKey)；下一个stage，在shuffle之后，做了一些transformation的操作

    如果Hive表对应了20个block，而我们自己设置的并行度是100，

    那么第一个stage的并行度是不受我们控制的，就只有20个task，第二个stage的才是我们设置的并行度100个

问题出在哪里了？

    SparkSQL 默认情况下，我们是没办法手动设置并行度的，所以可能造成问题，也可能不造成问题，

    SparkSQL后面的transformation算子操作，可能是很复杂的业务逻辑，甚至是很复杂的算法，

    如果SparkSQL默认的并行度设置的很少，20个，然后每个task要处理为数不少的数据量，

    还要执行很复杂的算法，这就导致第一个stage特别慢，第二个stage 1000个task，特别快！

解决办法：

    repartition：

        使用SparkSQL这一步的并行度和task的数量肯定是没办法改变了，但是可以将SparkSQL查出来的RDD，

        使用repartition算子进行重新分区，比如分多个partition，20 -> 100个；

        然后从repartition以后的RDD，并行度和task数量，就会按照我们预期的来了，

        就可以避免在跟SparkSQL绑定在一起的stage中的算子，只能使用少量的task去处理大量数据以及复杂的算法逻辑

5.算子操作reduceByKey：

reduceByKey相较于普通的shuffle操作(不如groupByKey)，他的一个特点就是会进行map端的本地聚合；

对map端给下个stage每个task创建的输出文件中，写数据之前，就会进行本地的combiner操作，也就是多每个key的value，都会执行算子函数(_+_)，减少了磁盘IO，较少了磁盘空间的占用,在reduce端的缓存也变少了

11.troubleshooting之控制reduce端缓冲大小以避免内存溢出(OOM)

new SparkConf().set("spark.reducer.maxSizeInFlight","24") //默认是48M

Map端的task是不断地输出数据的，数据量可能是很大的，

    但是其reduce端的task，并不是等到Map端task将属于自己的那个分数据全部写入磁盘后，再去拉取的

    Map端写一点数据，reduce端task就会去拉取一小部分数据，立刻进行后面的聚合，算子函数的应用；

    每次reduce能够拉取多少数据，是由reduce端buffer来定，因为拉取过来的数据都是放入buffer中的，

    然后采用后面的executor分配的堆内存占比(0.2),去进行后续的聚合，函数操作

reduce端buffer 可能会出现什么问题？

    reduce端buffer默认是48M，也许大多时候，还没有拉取满48M，也许是10M，就计算掉了，

    但是有时候，Map端的数据量特别大，写出的速度特别快，reduce端拉取的时候，全部到达了自己缓冲的最大极限48M，全部填满，

    这个时候，再加上reduce端执行的聚合函数代码，可能会创建大量的对象，也许一下子内存就撑不住了，就会造成OOM，reduce端的内存就会造成内存泄漏

如何解决？

    这个时候，我们应该减少reduce端task缓冲的大小，我们宁愿多拉取几次，但是每次同时能拉取到reduce端每个task的数据量比较少，就不容易发生OOM，比如调成12M；

    在实际生产中，这种问题是很常见的，这是典型的以性能换执行的原理，

    reduce的缓冲小了，不容易造成OOM了，但是性能一定是有所下降的，你要拉取的次数多了，

    就会走更多的网络IO流，这时候只能走牺牲性能的方式了；

曾经一个经验：

    曾经写了一个特别复杂的spark作业，写完代码后，半个月就是跑步起来，里面各种各样的问题，

    需要进行troubleshooting，调节了十几个参数，其中里面就有reduce端缓冲的大小，最后，

    总算跑起来了！

12. troubleshooting之解决JVM GC导致的shuffle拉取文件失败：

过程：

    第一个stage的task输出文件的同时 ，会像Driver上记录这些数据信息，然后下一个stage的task想要得到上个stage的数据，

    就得像Driver所要元数据信息，然后去像上一个的stage的task生成的文件中拉取数据。

问题场景：

    在spark作业中，有时候经常出现一种情况，就是log日志报出：shuffle file not found..,

    有时候他会偶尔出现一次，有的时候出现一次后重新提交stage、task，重新执行一遍 就好了。

分析问题：

    executor在JVM进程中，可能内存不太够用，那么此时就很可能执行GC，minor gc 或者 full gc，

    总之一旦发生gc后，就会导致所有工作线程全部停止，比如BlockManager，基于netty的网络通信。

    第二个stage的task去拉取数据的时候，上一个executor正好在进行gc，就导致拉取了半天也没拉取到数据，

    那为什么第二次提交stage的时候，就又可以了呢？

        因为第二次提交的时候，上一个executor已经完成了gc。

解决：

    spark.shuffle.io.maxRetries 3[默认3次]

        shuffle 文件拉取时，如果没有拉取到，最多或者重试几次，默认3次

    spark.shuffle.io.retryWait 5s [默认5s]

        每一次重新拉取文件的时间间隔，默认5s

    默认情况下，第一个stage的executor正在漫长的full gc，第二个stage的executor尝试去拉取数据，

    结果没拉取到，这样会反复重试拉取3次，中间间隔时间5s，也就是总共15s，拉取不成功，就报 shuffle file not found

        我们可以增大上面两个参数的值：

            spark.shuffle.io.maxRetries 60次

            spark.shuffle.io.retryWait 60s

            最多可以忍受一个小时没有拉取到shuffle file，这只是一个设置最大的可能值，

            full gc 也不可能一个小时都没结束把，

            这样就解决了因为gc 而无法拉取到数据的问题

13. troubleshooting之解决yarn-cluster模式的JVM栈内存溢出问题

yarn-cluster运行流程：

    1.本地机器执行spark-submit脚本[yarn-cluster模式]，提交spark application给resourceManager

    2. resourceManager找到一个节点[nodeManager]启动applicationMaster[Driver进程]

    3. applicationMaster找resourceManager申请executor

    4. resourceManager分配container(内存+cpu)

    5. applicationMaster找到对应nodeManager申请启动executor

    6. nodeManager启动executor

    7. executor找applicationMaster进行反向注册

    到这里为止，applicationMaster(Driver)就知道自己有哪些资源可以用(executor)，

    然后就会去执行job，拆分stage，提交stage的task，进行task调度，

    分配到各个executor上面去执行。

yarn-client 和 yarn-cluster的区别：

    yarn-client模式Driver运行在本地机器上；yarn-cluster模式Driver是运行在yarn集群上的某个nodeManager节点上的；

    yarn-client模式会导致本地机器负责spark作业的调用，所以网卡流量会激增，yarn-cluster没有这个问题；

    yarnclient的Driver运行在本地，通常来说本地机器和yarn集群都不会在一个机房，所以性能不是特别好；

    yarn-cluster模式下，Driver是跟yarn集群运行在一个机房内，性能上也会好很好；

实践经验碰到的yarn-cluster的问题：

    有时候运行了包含spark sql的spark作业，可能会遇到 在yarn-client上运行好好地，在yarn-cluster模式下，

    可能无法提交运行，会报出JVM的PermGen(永久代)的内存溢出-OOM；

    Yarn-client模式下，Driver是运行在本地机器的，spark使用的JVM的PerGen的配置，是本地的spark-class文件，

    (spark客户端是默认有配置的),JVM的永久代大小默认是128M，这个是没问题的；

    但是在Yarn-cluster模式下，Driver是运行在yarn集群的某个节点上的，使用的是没有经过配置的默认设置82M(PerGen永久代大小)

    spark sql内部会进行很负责的sl语义解析、语法树的转换，特别复杂，在这种情况下，如果sql特别复杂，

    很可能会导致性能的消耗，内存的消耗，可能对PermGen永久代的内存占比就很大

    所以此时，如果对PermGen的内存占比需求多与82M，但是又小于128M，就会出现类似上面的情况，

    yarn-client可以正常运行因为他的默认permgen大小是128M，但是yarn-cluster的默认是82M，就会出现PermGen OOM -- PermGen out of memory

解决：

    spark-submit脚本中加入参数：

        --conf spark.driver.extraJavaOptions='-XX:PermSize=128M -XX:MxPermSize=256M'

        这样就设置了永久代的大小默认128M，最大256M，那么这样的话，就可以保证spark作业不会出现上面的PermG