Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架，Spark，拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是——Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。
Spark 是一种与 Hadoop 相似的开源集群计算环境，但是两者之间还存在一些不同之处，这些有用的不同之处使 Spark 在某些工作负载方面表现得更加优越，换句话说，Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。
Spark 是在 Scala 语言中实现的，它将 Scala 用作其应用程序框架。与 Hadoop 不同，Spark 和 Scala 能够紧密集成，其中的 Scala 可以像操作本地集合对象一样轻松地操作分布式数据集。
学习整理：
-
spark为什么要划分DAG？
spark会根据宽依赖窄依赖来划分具体的stage，而依赖有2个作用：
- 其一用来解决数据容错的高效性；
- 其二用来划分stage。
RDD的依赖关系分为两种：窄依赖(Narrow Dependencies)与宽依赖(Wide Dependencies，源码中称为Shuffle Dependencies)
宽窄依赖
- 窄依赖：
每个父RDD的一个Partition最多被子RDD的一个Partition所使用（1:1 或 n:1）。例如map、filter、union等操作都会产生窄依赖；
子RDD分区通常对应常数个父RDD分区(O(1)，与数据规模无关）。- 宽依赖：一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖；（1:m 或 n:m）
子RDD分区通常对应所有的父RDD分区(O(n)，与数据规模有关）。
相比于宽依赖，窄依赖对优化很有利 ，主要基于以下两点：
- 宽依赖往往对应着shuffle操作，需要在运行过程中将同一个父RDD的分区传入到不同的子RDD分区中，中间可能涉及多个节点之间的数据传输；而窄依赖的每个父RDD的分区只会传入到一个子RDD分区中，通常可以在一个节点内完成转换。
- 当RDD分区丢失时（某个节点故障），spark会对数据进行重算。
- 对于窄依赖，由于父RDD的一个分区只对应一个子RDD分区，这样只需要重算和子RDD分区对应的父RDD分区即可，所以这个重算对数据的利用率是100%的；
对于宽依赖，重算的父RDD分区对应多个子RDD分区，这样实际上父RDD 中只有一部分的数据是被用于恢复这个丢失的子RDD分区的，另一部分对应子RDD的其它未丢失分区，这就造成了多余的计算；
更一般的，宽依赖中子RDD分区通常来自多个父RDD分区，极端情况下，所有的父RDD分区都要进行重新计算。
如下图所示，b1分区丢失，则需要重新计算a1,a2和a3，这就产生了冗余计算(a1,a2,a3中对应b2的数据)。
窄链接
区分这两种依赖很有用。
- 首先，窄依赖允许在一个集群节点上以流水线的方式（pipeline）计算所有父分区。例如，逐个元素地执行map、然后filter操作；而宽依赖则需要首先计算好所有父分区数据，然后在节点之间进行Shuffle，这与MapReduce类似。
- 第二，窄依赖能够更有效地进行失效节点的恢复，即只需重新计算丢失RDD分区的父分区，而且不同节点之间可以并行计算；而对于一个宽依赖关系的Lineage图，单个节点失效可能导致这个RDD的所有祖先丢失部分分区，因而需要整体重新计算。
【误解】之前一直理解错了，以为窄依赖中每个子RDD可能对应多个父RDD,当子RDD丢失时会导致多个父RDD进行重新计算，所以窄依赖不如宽依赖有优势。而实际上应该深入到分区级别去看待这个问题，而且重算的效用也不在于算的多少，而在于有多少是冗余的计算。窄依赖中需要重算的都是必须的，所以重算不冗余。
- 窄依赖的函数有：map、filter、union、join(父RDD是hash-partitioned )、mapPartitions、mapValues
- 宽依赖的函数有：groupByKey、join(父RDD不是hash-partitioned )、partitionBy
依赖的继承关系：
val rdd1 = sc.parallelize(1 to 10, 1)
val rdd2 = sc.parallelize(11 to 20, 1)
val rdd3 = rdd1.union(rdd2)
rdd3.dependencies.size// 长度为2，值为rdd1、rdd2，意为rdd3依赖rdd1、rdd2
rdd3.dependencies
rdd3.dependencies(0).rdd.collect// 打印rdd1的数据
rdd3.dependencies(1).rdd.collect// 打印rdd2的数据
rdd3.dependencies(3).rdd.collect// 数组越界，报错
哪些RDD Actions对应shuffleDependency？下面的join（r5）好像就没有shuffleDependency
val r1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"))
val r2 = r1.keyBy(_.length)
val r3 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"))
val r4 = r3.keyBy(_.length)
val r5 = r2.join(r4)
回答：
- join不一定会有shuffleDependency，上面的操作中就没有。
- redueceByKey会产生shuffleDependency。
- 注意上面操作中的keyBy，和我的想象不太一样。要注意一下。
- keyBy：与map操作较为类似，给每个元素增加了一个key
以下这个例子有点意思：
val r1 = sc.textFile("hdfs:///user/hadoop/data/block_test1.csv")
r1
val r2 = r1.dependencies(0).rdd
r2.partitions.size
r2.preferredLocations(r2.partitions(0))
r2.preferredLocations(r2.partitions(3))
有意思的地方在于（查找依赖、优先位置）：
- r1的类型为MapPartitionsRDD
- r1依赖于r2，如果没有这个赋值语句是看不出来的。r2的类型为：HadoopRDD
- 可以检索r2各个分区的位置，该hdfs文件系统的副本数设置为2
RDD的容错（lineage、checkpoint）
一般来说，分布式数据集的容错性有两种方式：数据检查点和记录数据的更新（CheckPoint Data，和Logging The Updates）。
面向大规模数据分析，数据检查点操作成本很高，需要通过数据中心的网络连接在机器之间复制庞大的数据集，而网络带宽往往比内存带宽低得多，同时还需要消耗更多的存储资源。
因此，Spark选择记录更新的方式。但是，如果更新粒度太细太多，那么记录更新成本也不低。因此，RDD只支持粗粒度转换，即只记录单个块上执行的单个操作（记录如何从其他RDD转换而来，即lineage），然后将创建RDD的一系列变换序列（每个RDD都包含了他是如何由其他RDD变换过来的以及如何重建某一块数据的信息。因此RDD的容错机制又称“血统(Lineage)”容错）记录下来，以便恢复丢失的分区。
Lineage本质上很类似于数据库中的重做日志（Redo Log），只不过这个重做日志粒度很大，是对全局数据做同样的重做进而恢复数据。
Lineage容错原理：
在容错机制中，如果一个节点死机了，而且运算窄依赖，则只要把丢失的父RDD分区重算即可，不依赖于其他节点。
而宽依赖需要父RDD的所有分区都存在，重算就很昂贵了。
可以这样理解开销的经济与否：
在窄依赖中，在子RDD的分区丢失、重算父RDD分区时，父RDD相应分区的所有数据都是子RDD分区的数据，并不存在冗余计算。
在宽依赖情况下，丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的，会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据，这样就会产生冗余计算开销，这也是宽依赖开销更大的原因。
因此如果使用Checkpoint算子来做检查点，不仅要考虑Lineage是否足够长，也要考虑是否有宽依赖，对宽依赖加Checkpoint是最物有所值的。
Checkpoint机制。在以下2种情况下，RDD需要加检查点：
- DAG中的Lineage过长，如果重算，则开销太大（如在多次迭代中）
- 在宽依赖上做Checkpoint获得的收益更大
由于RDD是只读的，所以Spark的RDD计算中一致性不是主要关心的内容，内存相对容易管理，这也是设计者很有远见的地方，这样减少了框架的复杂性，提升了性能和可扩展性，为以后上层框架的丰富奠定了强有力的基础。
在RDD计算中，通过检查点机制进行容错，传统做检查点有两种方式：通过冗余数据和日志记录更新操作。
在RDD中的doCheckPoint方法相当于通过冗余数据来缓存数据，而之前介绍的血统就是通过相当粗粒度的记录更新操作来实现容错的。
检查点（本质是通过将RDD写入Disk做检查点）是为了通过lineage做容错的辅助，Lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销。
checkpoint与cache的关系:
- 从本质上说：checkpoint是容错机制；cache是优化机制
- checkpoint将数据写到共享存储中（hdfs）；cache通常是内存中
- 运算时间很长或运算量太大才能得到的 RDD，computing chain 过长或依赖其他 RDD 很多的RDD，需要做checkpoint。会被重复使用的（但不能太大）RDD，做cache。实际上，将 ShuffleMapTask 的输出结果存放到本地磁盘也算是 checkpoint，只不过这个checkpoint 的主要目的是去 partition 输出数据。
- RDD 的checkpoint 操作完成后会斩断lineage，cache操作对lineage没有影响。
checkpoint 在 Spark Streaming中特别重要，spark streaming 中对于一些有状态的操作，这在某些 stateful 转换中是需要的，在这种转换中，生成 RDD 需要依赖前面的 batches，会导致依赖链随着时间而变长。为了避免这种没有尽头的变长，要定期将中间生成的 RDDs 保存到可靠存储来切断依赖链，必须隔一段时间进行一次checkpoint。
cache 和 checkpoint 是有显著区别的，缓存把 RDD 计算出来然后放在内存中， 但是RDD 的依赖链（相当于数据库中的redo 日志），也不能丢掉，当某个点某个 executor 宕了，上面cache 的RDD就会丢掉，需要通过依赖链重放计算出来，不同的是，checkpoint 是把 RDD 保存在 HDFS中，是多副本可靠存储，所以依赖链就可以丢掉了，即斩断了依赖链，是通过复制实现的高容错。
注意：checkpoint需要把 job 重新从头算一遍，最好先cache一下，checkpoint就可以直接保存缓存中的 RDD 了，就不需要重头计算一遍了，对性能有极大的提升。
checkpoint的使用与流程
checkpoint 的正确使用姿势:
val data = sc.textFile("/tmp/spark/1.data").cache() // 注意要cache
sc.setCheckpointDir("/tmp/spark/checkpoint")
data.checkpoint
data.count
//问题：cache和checkpoint有没有先后的问题；有了cache可以避免第二次计算，我在代码中可以看见相关的说明！！！
使用很简单， 就是设置一下 checkpoint 目录，然后再rdd上调用 checkpoint 方法， action 的时候就对数据进行了 checkpoint
checkpoint 写流程:
RDD checkpoint 过程中会经过以下几个状态，
[ Initialized –> marked for checkpointing –> checkpointing in progress –> checkpointed ]
参见：
Spark作业调度中stage的划分：https://wongxingjun.github.io/2015/05/25/Spark%E4%BD%9C%E4%B8%9A%E8%B0%83%E5%BA%A6%E4%B8%ADstage%E7%9A%84%E5%88%92%E5%88%86/
Hadoop 和 Spark 的关系：
https://www.cnblogs.com/ITtangtang/p/7967886.html
Spark Shuffle原理及相关调优：
http://sharkdtu.com/posts/spark-shuffle.html
Distributed System | Spark RDD 论文总结：
https://juejin.im/entry/58b7e625a22b9d005ed08f7a
spark的宽依赖窄依赖:
https://cloud.tencent.com/info/a772127faac92aa56593190371f76d8c.html
-
产生数据倾斜的原因？
- 从程序上来看
由于countdistinct、group by、join等操作，这些都会触发Shuffle动作，一旦触发，所有相同key的值就会拉到一个或几个节点上，此时如果某个key对应的数据量特别大的话，就会发生数据倾斜。 - 从数据上来看
两个不同的人开发的数据表，如果我们的数据规范不太完善的话，会出现一种情况，就是某一张表的的key存在大量的null值或相同的一个数据，则会在数据分布的时候倾斜。 - 从业务上来看
数据的分布往往会与业务有关系，比如各城市订单系统，如果某城市做了大量宣传，随着订单量的爆发式增长，当使用groupby的时候就会单节点大量数据堆积，数据倾斜。
参见：
漫谈千亿级数据优化实践：数据倾斜（纯干货）：https://segmentfault.com/a/1190000009166436
Hive - hive.groupby.skewindata环境变量与负载均衡：
https://blog.csdn.net/zj360202/article/details/38420575
-
数据倾斜有哪些现象？
数据倾斜发生后现现象会有两种情况：
- 大部份的Task执行的时候会很快，当发生数据倾斜后的task会执行很长时间。
- 有时候数据倾斜直接报OOM即：JVM Out Of Memory内存溢出的错误。
参见：
spark数据倾斜与解决方法：https://www.cnblogs.com/runnerjack/p/8258291.html
-
如何定位数据倾斜代码段？
数据倾斜多发生在shuffle阶段，我们需要留意会触发shuffle操作的算子，如：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition。具体到不同的情况有：
- 某个task执行特别慢
第一步，我们要确定具体是哪个stage导致了运行的缓慢，通过查看log或者Spark Web UI查看stage中task分配的数据量，便可以确定；
第二步，找到运行缓慢的stage之后，我们就可以根据stage划分的原理推断出stage对应代码在哪一部分（注意分析代码中引起shuffle操作的算子）。 - 某个task莫名其妙内存溢出
第一步，我们可以到log中找到异常栈，异常栈一般会标明程序出错的代码位置。
第二步，在内存溢出代码位置附近找会触发shuffle操作的算子，在自己程序没有bug的前提下，很有可能就是该算子造成数据溢出。
需要注意的是，不能说内存溢出就是数据倾斜，因为无法排除程序bug，或偶然的数据异常导致的溢出，建议还是按照1.中所讲，结合stage中task分配的数据量来最终确认是否为数据溢出。
参考：
Spark性能优化指南——高级篇：
https://tech.meituan.com/spark-tuning-pro.html
Spark函数详解系列之RDD基本转换：https://www.cnblogs.com/MOBIN/p/5373256.html
-
讲讲如何优化spark？
spark优化方案：开发调优、资源调优、数据倾斜调优、shuffle调优。
- 开发调优
开发Spark作业的过程中注意和应用一些性能优化的基本原则，包括RDD lineage设计、算子的合理使用、特殊操作的优化等。
原则一：避免创建重复的RDD
对于同一份数据，只应该创建一个RDD，不能创建多个RDD来代表同一份数据。原则二：尽可能复用同一个RDD
多个RDD的数据有重叠或者包含的情况，我们应该尽量复用一个RDD，这样可以尽可能地减少RDD的数量，从而尽可能减少算子执行的次数。原则三：对多次使用的RDD进行持久化
对多次使用的RDD进行持久化（调用cache()和persist()），RDD进行算子操作时，都会直接从内存或磁盘中提取持久化的RDD数据，然后执行算子，而不会从源头处重新计算一遍这个RDD，再执行算子操作。
spark持久化级别
建议选择顺序为：MEMORY_ONLY > MEMORY_ONLY_SER > MEMORY_AND_DISK_SER
不建议：MEMORY_AND_DISK、DISK_ONLY、后缀为_2的级别。原则四：尽量避免使用shuffle类算子
我们的开发过程中，能避免则尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子，尽量使用map类的非shuffle算子，以避免多余的磁盘IO和网络数据传输。原则五：使用map-side预聚合的shuffle操作
因为业务需要，一定要使用shuffle操作，无法用map类的算子来替代，那么尽量使用可以map-side预聚合的算子（如reduceByKey代替aggregateByKey），map-side预聚合使每个节点本地对相同的key先进行一次聚合操作，类似于MapReduce中的本地combiner。原则六：使用高性能的算子
除了shuffle相关的算子有优化原则之外，其他的算子也都有着相应的优化原则。
- 使用reduceByKey/aggregateByKey替代groupByKey
- 使用mapPartitions替代普通map
- 使用foreachPartitions替代foreach
- 使用filter之后进行coalesce操作
- 使用repartitionAndSortWithinPartitions替代repartition与sort类操作
原则七：广播大变量
有时在开发过程中，会遇到需要在算子函数中使用外部变量的场景（尤其是大变量，比如100M以上的大集合），那么此时就应该使用Spark的广播（Broadcast）功能来提升性能。广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本。（若不广播，一task一副本）原则八：使用Kryo优化序列化性能
在Spark中，主要有三个地方涉及到了序列化：
- 在算子函数中使用到外部变量时，该变量会被序列化后进行网络传输。
- 将自定义的类型作为RDD的泛型类型时（比如JavaRDD，Student是自定义类型），所有自定义类型对象，都会进行序列化。因此这种情况下，也要求自定义的类必须实现Serializable接口。
- 使用可序列化的持久化策略时（比如MEMORY_ONLY_SER），Spark会将RDD中的每个partition都序列化成一个大的字节数组。
通过使用Kryo序列化类库，序列化和反序列化的性能得到优化，但要注意Kryo最好要注册所有需要进行序列化的自定义类型。
原则九：优化数据结构
Java中，有三种类型比较耗费内存：
- 对象，每个Java对象都有对象头、引用等额外的信息，因此比较占用内存空间。
- 字符串，每个字符串内部都有一个字符数组以及长度等额外信息。
- 集合类型，比如HashMap、LinkedList等，因为集合类型内部通常会使用一些内部类来封装集合元素，比如Map.Entry。
建议在适的情况下，使用占用内存较少的数据结构，但是前提是要保证代码的可维护性
-
资源调优
资源参数设置的不合理，可能会导致没有充分利用集群资源，作业运行会极其缓慢；或者设置的资源过大，队列没有足够的资源来提供，进而导致各种异常。因此我们必须对Spark作业的资源使用原理有一个清晰的认识，并知道在Spark作业运行过程中，有哪些资源参数是可以设置的，以及如何设置合适的参数值。
Spark作业基本运行原理
可以看到Executor的内存主要分为三块：
第一块是让task执行我们自己编写的代码时使用，默认是占Executor总内存的20%；
第二块是让task通过shuffle过程拉取了上一个stage的task的输出后，进行聚合等操作时使用，默认也是占Executor总内存的20%；
第三块是让RDD持久化时使用，默认占Executor总内存的60%。task的执行速度是跟每个Executor进程的CPU core数量有直接关系的。一个CPU core同一时间只能执行一个线程。而每个Executor进程上分配到的多个task，都是以每个task一条线程的方式，多线程并发运行的。如果CPU core数量比较充足，而且分配到的task数量比较合理，那么通常来说，可以比较快速和高效地执行完这些task线程。
具体参数：
- num-executors
设置Spark作业总共要用多少个Executor进程来执行，一般设置50~100个为宜。- executor-memory
设置每个Executor进程的内存，JVM OOM异常有关，4G ~ 8G较为合适；若为共享资源队列，num-executors乘以executor-memory最好控制在资源队列最大总内存的1/3 ~ 1/2。- executor-cores
设置Executor进程的CPU core数量，决定了每个Executor进程并行执行task线程的能力，以2 ~ 4为宜，同样num-executors * executor-cores不要超过队列总CPU core的1/3 ~ 1/2- driver-memory
设置Driver进程的内存，一般不设置或者设置1G就够了，若需要使用collect算子，需要设置合适的值来避免OOM内存溢出的问题。- spark.default.parallelism
设置每个stage的默认task数量，默认task数量为500~1000为宜，官方建议设置num-executors * executor-cores的2~3倍比较合适。- spark.storage.memoryFraction
RDD持久化数据在Executor内存中能占的比例，默认是0.6。RDD持久化多，则适当提高；shuffle操作多，持久化少，则适当减少。当出现用户代码的内存不够用时，也建议调低该值。- spark.shuffle.memoryFraction
设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例，默认是0.2。同样，shuffle操作多时建议提高该值，否则溢出部分写磁盘影响性能。资源调优没有固定值，具体问题具体分析，注意结合Spark作业中的shuffle操作数量、RDD持久化操作数量以及spark web ui中显示的作业gc情况。
附实列：
./bin/spark-submit \ --master yarn-cluster \ --num-executors 100 \ --executor-memory 6G \ --executor-cores 4 \ --driver-memory 1G \ --conf spark.default.parallelism=1000 \ --conf spark.storage.memoryFraction=0.5 \ --conf spark.shuffle.memoryFraction=0.3 \
- 数据倾斜调优
数据倾斜的原理很简单：在进行shuffle的时候，必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理，比如按照key进行聚合或join等操作。此时如果某个key对应的数据量特别大的话，就会发生数据倾斜。
数据倾斜的解决方案:
解决方案一：使用Hive ETL预处理数据
适用于导致数据倾斜的是Hive表。hive对数据预先进行聚合或join操作，spark不再需要shuffle，所以完全规避掉了数据倾斜，Spark作业的性能大幅度提升。但治标不治本，Hive ETL中还是会发生数据倾斜。解决方案二：过滤少数导致倾斜的key
如果发现导致倾斜的key就少数几个，而且对计算本身的影响并不大的话，那么干脆就直接过滤掉那少数几个key，结合sample取样还可以实现动态判定。但大多数情况下，导致倾斜的key还是很多的，并不是只有少数几个。解决方案三：提高shuffle操作的并行度
增加shuffle read task的数量（spark.sql.shuffle.partitions），可以让原本分配给一个task的多个key分配给多个task，从而让每个task处理比原来更少的数据。缓解了数据倾斜，但没有彻底根除问题，一般作为应急手段。
提高shuffle并行度解决方案四：两阶段聚合（局部聚合+全局聚合）
适用于聚合类的shuffle操作导致的数据倾斜。将原本相同的key通过附加随机前缀的方式，变成多个不同的key，让原本被一个task处理的数据分散到多个task上去做局部聚合，接着去除掉随机前缀，再次进行全局聚合。解决单个task处理数据量过多的问题，但如果是join类的shuffle操作，还得用其他的解决方案。
局部聚合+全局聚合解决方案五：将reduce join转为map join
当join操作中的一个RDD或表的数据量比较小（比如几百M或者一两G）时，可以采用广播小RDD全量数据+map算子来实现与join同样的效果，也就是map join，此时就不会发生shuffle操作，也就不会发生数据倾斜。对join操作导致的数据倾斜，效果非常好，但只适用于一个大表和一个小表的情况。
reduce join 转 map join解决方案六：采样倾斜key并分拆join操作
适用于两个RDD/Hive表进行join的时候，两张表数据量都比较大，且出现数据倾斜，是因为其中某一个RDD/Hive表中的少数几个key的数据量过大，而另一个RDD/Hive表中的所有key都分布比较均匀。
实现思路：
- 对包含少数几个数据量过大的key的那个RDD，通过sample算子采样出一份样本来，然后统计一下每个key的数量，计算出来数据量最大的是哪几个key。
- 然后将这几个key对应的数据从原来的RDD中拆分出来，形成一个单独的RDD，并给每个key都打上n以内的随机数作为前缀，而不会导致倾斜的大部分key形成另外一个RDD。
- 接着将需要join的另一个RDD，也过滤出来那几个倾斜key对应的数据并形成一个单独的RDD，将每条数据膨胀成n条数据，这n条数据都按顺序附加一个0~n的前缀，不会导致倾斜的大部分key也形成另外一个RDD。
- 再将附加了随机前缀的独立RDD与另一个膨胀n倍的独立RDD进行join，此时就可以将原先相同的key打散成n份，分散到多个task中去进行join了。
而另外两个普通的RDD就照常join即可。
采样key并拆分join
该方案能有效解决数据倾斜，但如果导致倾斜的key特别多的话，那么这种方式也不适合。
解决方案七：使用随机前缀和扩容RDD进行join
如果在进行join操作时，RDD中有大量的key导致数据倾斜，那么进行分拆key也没什么意义，此时就只能使用最后一种方案了
具体实现：
- 该方案的实现思路基本和“解决方案六”类似，首先查看RDD/Hive表中的数据分布情况，找到那个造成数据倾斜的RDD/Hive表，比如有多个key都对应了超过1万条数据。
- 然后将该RDD的每条数据都打上一个n以内的随机前缀。
- 同时对另外一个正常的RDD进行扩容，将每条数据都扩容成n条数据，扩容出来的每条数据都依次打上一个0~n的前缀。
- 最后将两个处理后的RDD进行join即可。
该方案对join类型的数据倾斜基本都可以处理，而且效果也相对比较显著，但更多的是缓解数据倾斜，而不是彻底避免数据倾斜。而且需要对整个RDD进行扩容，对内存资源要求很高
解决方案八：多种方案组合使用
当遇到一个较为复杂的数据倾斜场景，那么可能需要将多种方案组合起来使用。
- 针对出现了多个数据倾斜环节的Spark作业，可以先运用解决方案一和二，预处理一部分数据，并过滤一部分数据来缓解；
- 其次可以对某些shuffle操作提升并行度，优化其性能；
- 最后还可以针对不同的聚合或join操作，选择一种方案来优化其性能。
4.shuffle调优
大多数Spark作业的性能主要就是消耗在了shuffle环节，因为该环节包含了大量的磁盘IO、序列化、网络数据传输等操作。因此，如果要让作业的性能更上一层楼，就有必要对shuffle过程进行调优。
先看看spark提供的shuffle运行原理：
未经优化的HashShuffleManager
未优化HashShuffleManager
优化后的HashShuffleManager
优化HashShuffleManager
SortShuffleManager 普通运行机制
SortShuffleManager 普通机制
SortShuffleManager bypass运行机制
- shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。
不是聚合类的shuffle算子。
SortShuffleManager bypass机制
具体参数：
- spark.shuffle.file.buffer
用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小，默认为32k，如果内存资源较为充足的话，可以适当增加这个参数的大小（比如64k），从而减少shuffle write过程中溢写磁盘IO次数。- spark.reducer.maxSizeInFlight
用于设置shuffle read task的buffer缓冲大小，而这个buffer缓冲决定了每次能够拉取多少数据。默认为48m，如果内存资源较为充足的话，可以适当增加这个参数的大小（比如96m），从而减少拉取数据的网络传输的次数。- spark.shuffle.io.maxRetries
设置shuffle read task从shuffle write task所在节点拉取数据时拉取失败后的最大重试次数。默认为3，对于有suffle耗时比较多的作业，建议增加重试最大次数（比如60次）。- spark.shuffle.io.retryWait
每次重试拉取数据的等待间隔，默认是5s,建议加大间隔时长（比如60s），以增加shuffle操作的稳定性。- spark.shuffle.memoryFraction
代表了Executor内存中，分配给shuffle read task进行聚合操作的内存比例，默认是20%（0.2），如果内存充足，而且很少使用持久化操作，建议调高这个比例。- spark.shuffle.manager（现已没有，默认为sort）
用于设置ShuffleManager的类型，hash、sort和tungsten-sort，现已统一为sort。- spark.shuffle.sort.bypassMergeThreshold
当shuffle read task的数量小于这个阈值（默认是200）时,启用SortShuffleManager bypass运行机制，如果的确不需要排序操作，那么建议将这个参数调大一些。- spark.shuffle.consolidateFiles(现已没有)
默认为false，如果使用HashShuffleManager，设置为true，那么就会开启consolidate机制，会大幅度合并shuffle write的输出文件。但因新版本统一使用SortShuffleManager，故该参数已无效。
最后必须提醒大家的是，影响一个Spark作业性能的因素，主要还是代码开发、资源参数以及数据倾斜，shuffle调优只能在整个Spark的性能调优中占到一小部分而已。
参见：
Spark性能优化指南——基础篇：
https://tech.meituan.com/spark-tuning-basic.html
Spark性能优化指南——高级篇：
https://tech.meituan.com/spark-tuning-pro.html
Spark Configuration:
https://spark.apache.org/docs/2.3.0/configuration.html
-
spark调度各调度算法理解？
先来看spark运行原理（yarn-cluster为例）。
如图所示在yarn-cluster模式下，提交一个Spark应用程序，首先通过Client向ResourceManager请求启动一个Application，同时检查是否有足够的资源满足Application的需求，如果资源条件满足，则准备ApplicationMaster的启动上下文，交给ResourceManager，并循环监控Application状态。
当提交的资源队列中有资源时，ResourceManager会在某个NodeManager上启动ApplicationMaster进程，ApplicationMaster会单独启动Driver后台线程，当Driver启动后，ApplicationMaster会通过本地的RPC连接Driver，并开始向ResourceManager申请Container资源运行Executor进程（一个Executor对应与一个Container），当ResourceManager返回Container资源，则在对应的Container上启动Executor。
Driver线程主要是初始化SparkContext对象，准备运行所需的上下文，然后一方面保持与ApplicationMaster的RPC连接，通过ApplicationMaster申请资源，另一方面根据用户业务逻辑开始调度任务，将任务下发到已有的空闲Executor上。
当ResourceManager向ApplicationMaster返回Container资源时，ApplicationMaster就尝试在对应的Container上启动Executor进程，Executor进程起来后，会向Driver注册，注册成功后保持与Driver的心跳，同时等待Driver分发任务，当分发的任务执行完毕后，将任务状态上报给Driver。
为了理解整个系统的调度，结合spark的运行原理，我们将调度分成yarn调度和spark调度两个部分，分别用于处理yarn上 不同应用间（spark和其他应用） 和 spark应用相关 的调度问题。
-------------------------------------A. YARN调度-------------------------------------
通过图我们看到完整的yarn运行原理，其中负责给应用分配资源的就是Scheduler，它在yarn资源调度中起到了非常重要的作用。
Yarn中有三种调度器可以选择：FIFO Scheduler ，Capacity Scheduler，Fair Scheduler
- FIFO Scheduler
FIFO Scheduler是最简单也是最容易理解的调度器，也不需要任何配置，但它并不适用于共享集群。大的应用可能会占用所有集群资源，这就导致其它应用被阻塞。
- Capacity Scheduler（容器调度器）
Capacity 调度器通过为每个组织分配专门的队列，然后再为每个队列分配一定的集群资源实现。队列内部还可以垂直划分子队列，子队列内任务的资源调度最终采用先进先出(FIFO)策略。另外Capacity调度器仍可能分配额外的资源给队列，即“弹性队列”(queue elasticity)的概念。
配置如下：
conf/yarn-site.xml<property> <name>yarn.resourcemanager.scheduler.class</name> <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value> </property>conf/capacity-scheduler.xml
<property> <name>yarn.scheduler.capacity.root.queues</name> <value>a,b</value> </property> <property> <name>yarn.scheduler.capacity.root.a.queues</name> <value>c,d</value> </property> <property> <name>yarn.scheduler.capacity.root.a.capacity</name> <value>40</value> <description>root下分配给a的比例</description> </property> <property> <name>yarn.scheduler.capacity.root.a.maximum-capacity</name> <value>50</value> <description>root下弹性分配给a的最大比例</description> </property>
- 队列设置：
通过mapreduce.job.queuename属性指定要用的队列,如果我们没有定义任何队列，所有的应用将会放在一个default队列中。
- Fair Scheduler（公平调度器）
Fair调度器的设计目标是为所有的应用分配公平的资源（对公平的定义可以通过参数来设置）。在上面的“Yarn调度器对比图”展示了一个队列中两个应用的公平调度；当然，公平调度在也可以在多个队列间工作。
Fair Scheduler
配置如下：
conf/yarn-site.xml<property> <name>yarn.resourcemanager.scheduler.class</name> <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value> </property> <property> <name>yarn.scheduler.fair.allocation.file</name> <value>conf/fair-scheduler.xml</value> <description>修改fair配置文件路径</description> </property>conf/fair-scheduler.xml
<allocations> <defaultQueueSchedulingPolicy>fair</defaultQueueSchedulingPolicy> <queue name="A"> <minResources>368640 mb,90 vcores</minResources> <maxResources>2334720 mb,570 vcores</maxResources> <maxRunningApps>70</maxRunningApps> <weight>5</weight> <schedulingPolicy>FIFO</schedulingPolicy> <queue name="B"> <minResources>122880 mb,30 vcores</minResources> <maxResources>1966080 mb,480 vcores</maxResources> <maxRunningApps>20</maxRunningApps> <weight>8</weight> </queue> </queue> <queuePlacementPolicy> <rule name="specified" create="false"/> <rule name="primaryGroup" create="false"/> <rule name="default" queue="A.B"/> </queuePlacementPolicy> </allocations>每个队列内部仍可以有不同的调度策略。队列的默认调度策略可以通过顶级元素<defaultQueueSchedulingPolicy>进行配置，如果没有配置，默认采用公平调度。尽管是Fair调度器，其仍支持在队列级别进行FIFO调度。每个队列的调度策略可以被其内部的<schedulingPolicy> 元素覆盖。
- 队列设置：
Fair调度器采用了一套基于规则的系统来确定应用应该放到哪个队列。<queuePlacementPolicy> 元素定义了一个规则列表，其中的每个规则会被逐个尝试直到匹配成功。<queuePlacementPolicy> <rule name="specified" create="false"/> <rule name="user" create="false"/> <rule name="default" /> </queuePlacementPolicy>上例第一个规则specified，则会把应用放到它指定的队列中，若这个应用没有指定队列名或队列名不存在，则说明不匹配这个规则，然后尝试下一个规则。user规则会尝试把应用放在以用户名命名的队列中，如果没有这个队列，不创建队列转而尝试下一个规则。当前面所有规则不满足时，则触发default规则。
- 抢占（Preemption）
抢占就是允许调度器杀掉占用超过其应占份额资源队列的containers，这些containers资源便可被分配到应该享有这些份额资源的队列中。需要注意抢占会降低集群的执行效率，因为被终止的containers需要被重新执行。可以通过设置一个全局的参数yarn.scheduler.fair.preemption=true来启用抢占功能。此外，还有两个参数用来控制抢占的过期时间（这两个参数默认没有配置，需要至少配置一个来允许抢占Container）：
- minimum share preemption timeout - fair share preemption timeout
- 如果队列在minimum share preemption timeout指定的时间内未获得最小的资源保障，调度器就会抢占containers。我们可以通过配置文件中的顶级元素<defaultMinSharePreemptionTimeout>为所有队列配置这个超时时间；我们还可以在<queue>元素内配置<minSharePreemptionTimeout>元素来为某个队列指定超时时间。
- 与之类似，如果队列在fair share preemption timeout指定时间内未获得平等的资源的一半（这个比例可以配置），调度器则会进行抢占containers。这个超时时间可以通过顶级元素<defaultFairSharePreemptionTimeout>和元素级元素<fairSharePreemptionTimeout>分别配置所有队列和某个队列的超时时间。上面提到的比例可以通过<defaultFairSharePreemptionThreshold>(配置所有队列)和<fairSharePreemptionThreshold>(配置某个队列)进行配置，默认是0.5。
------------------------------------B. SPARK调度------------------------------------
spark调度可划分Spark应用间资源调度，和Spark应用内资源调度两层。
- Spark应用之间的资源调度
资源静态分配：
- Standalone mode
独立部署的集群中都会以FIFO（first-in-first-out）模式顺序提交运行；
设置spark.cores.max或者spark.deploy.defaultCores 来限制单个应用所占用的节点个数；
设置spark.executor.memory 来控制各个应用的内存占用量。- Mesos
设置spark.mesos.coarse 设为true；
spark.cores.max 来控制各个应用的CPU总数；
spark.executor.memory 来控制各个应用的内存占用。- YARN
–num-executors 选项来控制Spark应用在集群中分配的执行器的个数;
–executor-memory 和 –executor-cores 来控制单个执行器（executor）所占用的资源。动态资源分配：
- 配置和部署
设置 spark.dynamicAllocation.enabled 为 true;
每个节点上启动一个外部混洗服务（external shuffle service），并在你的应用中将 spark.shuffle.service.enabled 设为true;
- Standalone mode,worker启动前设置 spark.shuffle.server.enabled 为true即可
- Mesos粗粒度模式,个节点上运行 ${SPARK_HOME}/sbin/start-mesos-shuffle-service.sh 并设置 spark.shuffle.service.enabled 为true;
- YARN,spark-<version>-yarn-shuffle.jar添加到NodeManager的classpath路径中;
配置yarn-site.xml，将 spark_shuffle 添加到 yarn.nodemanager.aux-services 中，然后将 yarn.nodemanager.aux-services.spark_shuffle.class 设为 org.apache.spark.network.yarn.YarnShuffleService，并将 spark.shuffle.service.enabled 设为 true；
重启各节点上的NodeManager；
- 资源分配策略
请求策略
Spark会分轮次来申请执行器。实际的资源申请，会在任务挂起 spark.dynamicAllocation.schedulerBacklogTimeout 秒后首次触发，其后如果等待队列中仍有挂起的任务，则每过 spark.dynamicAlloction.sustainedSchedulerBacklogTimeout 秒触发一次资源申请。另外，每一轮所申请的执行器个数以指数形式增长(如：1，2，4，8)。
移除策略
Spark应用会在某个执行器空闲超过 spark.dynamicAllocation.executorIdleTimeout 秒后将其删除。在绝大多数情况下，执行器的移除条件和申请条件都是互斥的，也就是说，执行器在有待执行任务挂起时，不应该空闲。
优雅地关闭执行器
为解决executor移除带来的中间数据丢失这一问题，就需要用到一个外部混洗服务（external shuffle service）用于存放中间数据。一旦该服务启用，Spark执行器不再从各个执行器上获取shuffle文件，转而从这个service获取。Mesos动态共享CPU
在这种模式下，每个Spark应用的内存占用仍然是固定且独占的（仍由spark.executor.memory决定），但是如果该Spark应用没有在某个机器上执行任务的话，那么其他应用可以占用该机器上的CPU。这种模式适用于集群中有大量不是很活跃应用的场景，但不适用低延迟的场景，因为当Spark应用需要使用CPU的时候，可能需要等待一段时间才能取得CPU的使用权。
- Yarn调度：
其作用是分配container，
- Spark应用内部的资源调度
在讲Spark内部资源调度前，我们来看看单个spark应用的时序图：
从上述时序图可知，Client只管提交Application并监控Application的状态。对于Spark的任务调度主要是集中在两个方面: 资源申请和任务分发，其主要是通过ApplicationMaster、Driver以及Executor之间来完成，下面详细剖析Spark任务调度每个细节。
spark调度涉及job、stage、task三个概念：
- Job是以Action方法为界，遇到一个Action方法则触发一个Job；
- Stage是Job的子集，以RDD宽依赖(即Shuffle)为界，遇到Shuffle做一次划分；
- Task是Stage的子集，以并行度(分区数)来衡量，分区数是多少，则有多少个task。
任务调度分stage调度和task调度两路，且有如下结构：
Spark RDD通过其Transactions操作，形成了RDD血缘关系图，即DAG，最后通过Action的调用，触发Job并调度执行。DAGScheduler负责Stage级的调度，主要是将DAG切分成若干Stages，并将每个Stage打包成TaskSet交给TaskScheduler调度。TaskScheduler负责Task级的调度，将DAGScheduler给过来的TaskSet按照指定的调度策略分发到Executor上执行，调度过程中SchedulerBackend负责提供可用资源，其中SchedulerBackend有多种实现，分别对接不同的资源管理系统（standalone、yarn或mesos）。
- Stage级的调度
Job由最终的RDD和Action方法封装而成，SparkContext将Job交给DAGScheduler提交，它会根据RDD的血缘关系构成的DAG进行切分，将一个Job划分为若干Stages（ShuffleMapStage和ResultStage），先提交父stage后提交子stage并监控相关状态信息。
- Task级的调度
Spark Task的调度是由TaskScheduler来完成，由前文可知，DAGScheduler将Stage打包到TaskSet交给TaskScheduler，TaskScheduler会将其封装为TaskSetManager加入到调度队列中，TaskSetManager负责监控管理同一个Stage中的Tasks，TaskScheduler就是以TaskSetManager为单元来调度任务。前面也提到，TaskScheduler初始化后会启动SchedulerBackend，它负责跟外界打交道，接收Executor的注册信息，并维护Executor的状态，所以说SchedulerBackend是管“粮食”的，同时它在启动后会定期地去“询问”TaskScheduler有没有任务要运行，也就是说，它会定期地“问”TaskScheduler“我有这么余量，你要不要啊”，TaskScheduler在SchedulerBackend“问”它的时候，会从调度队列中按照指定的调度策略选择TaskSetManager去调度运行。
调用流程
调度策略
TaskScheduler支持两种调度策略，一种是FIFO，也是默认的调度策略，另一种是FAIR。在TaskScheduler初始化过程中会实例化rootPool，表示树的根节点，是Pool类型。
类型关系
Pool为中间节点，TaskSetManager为叶子节点。
如果是采用FIFO调度策略，则直接简单地将TaskSetManager按照先来先到的方式入队，出队时直接拿出最先进队的TaskSetManager。如图所示，TaskSetManager保存在一个FIFO队列中。
FIFO
其调度算法FIFOSchedulingAlgorithm只是比较job ID和stage ID。
- 使用FAIR调度策略时
应用程序中使用两个线程分别调用Action方法，即有两个Job会并发提交，这两个Job被切分成若干TaskSet后放入指定pool，其调度树大致如下图所示。
FAIR
在出队时，则会对所有TaskSetManager排序，具体排序过程是从根节点rootPool开始，递归地去排序子节点（即pool，非叶子节点），最后合并到一个ArrayBuffer。其算法FairSchedulingAlgorithm则复杂一些根据runningTasks、minshare、weight来综合判断优先级。
比较顺序和权重优先级：
- A,B中只有一个runningTasks<minShare，则满足该情况优先；
- A,B同时runningTasks>minShare或runningTasks<minShare的情况下: runningTasks/minShare > runningTasks/weight > Schedulable.name(实在不行就比较名字)
配置文件：
{spark_base_dir}/conf/spark_default.conf：spark.scheduler.mode FAIR设置命令:
sc.setLocalProperty("spark.scheduler.pool", "pool1") //pool指定 sc.setLocalProperty("spark.scheduler.pool", null) //取消pool指定可为以线程为单位为driver（SparkContext）分别设置pool，该driver下的所有job使用该资源池。
配置资源池：
conf.set("spark.scheduler.allocation.file", "/path/to/file")配置文件格式：
<?xml version="1.0"?> <allocations> <pool name="production"> <schedulingMode>FAIR</schedulingMode> <weight>1</weight> <minShare>2</minShare> </pool> <pool name="test"> <schedulingMode>FIFO</schedulingMode> <weight>2</weight> <minShare>3</minShare> </pool> </allocations>完整可参考conf/fairscheduler.xml.template，默认配置（schedulingMode：FIFO，weight：1，minShare：0）
注意：对于pool调度是fair，但对于同一pool中job（实际为job中stage对应TaskSetManager）则是采用FIFO。
- 本地化调度
从调度队列中拿到TaskSetManager后，那么接下来的工作就是TaskSetManager按照一定的规则一个个取出Task给TaskScheduler，TaskScheduler再交给SchedulerBackend去发到Executor上执行。
在TaskSetManager初始化过程中，会对Tasks按照Locality级别进行分类，Task的Locality有五种，优先级由高到低顺序：PROCESS_LOCAL(指定的Executor)，NODE_LOCAL(指定的主机节点)，NO_PREF(无所谓)，RACK_LOCAL(指定的机架)，ANY(满足不了Task的Locality就随便调度)。这五种Locality级别存在包含关系，RACK_LOCAL包含NODE_LOCAL，NODE_LOCAL包含PROCESS_LOCAL，然而ANY包含其他所有四种, 这样调度执行时，满足不了PROCESS_LOCAL，就逐步退化到NODE_LOCAL，RACK_LOCAL，ANY。
本地化调度
TaskSetManager在决定调度哪些Task时，是通过上面流程图中的resourceOffer方法来实现，为了尽可能地将Task调度到它的preferredLocations上，它采用一种延迟调度算法，实现如下：
def resourceOffer( execId: String, //调度任务的Executor Id host: String, //主机地址 maxLocality: TaskLocality.TaskLocality) //最大可容忍的Locality级别 : Option[TaskDescription]本地化调度流程
失败重试与黑名单机制
TaskSetManager在记录Task失败次数过程中，会记录它上一次失败所在的Executor Id和Host，这样下次再调度这个Task时，会使用黑名单机制，避免它被调度到上一次失败的节点上，起到一定的容错作用。黑名单记录Task上一次失败所在的Executor Id和Host，以及其对应的“黑暗”时间，“黑暗”时间是指这段时间内不要再往这个节点上调度这个Task了。推测式执行
TaskScheduler在启动SchedulerBackend后，还会启动一个后台线程专门负责推测任务的调度，推测任务是指对一个Task在不同的Executor上启动多个实例，如果有Task实例运行成功，则会干掉其他Executor上运行的实例。推测调度线程会每隔固定时间检查是否有Task需要推测执行，如果有，则会调用SchedulerBackend的reviveOffers去尝试拿资源运行推测任务。
推测式执行
检查是否有Task需要推测执行的逻辑最后会交到TaskSetManager，TaskSetManager采用基于统计的算法，检查Task是否需要推测执行，算法流程大致如下图所示。
推测式执行
TaskSetManager首先会统计成功的Task数，当成功的Task数超过75%(可通过参数spark.speculation.quantile控制)时，再统计所有成功的Tasks的运行时间，得到一个中位数，用这个中位数乘以1.5(可通过参数spark.speculation.multiplier控制)得到运行时间门限，如果在运行的Tasks的运行时间超过这个门限，则对它启用推测。算法逻辑较为简单，其实就是对那些拖慢整体进度的Tasks启用推测，以加速整个TaskSet即Stage的运行。
参见：
Spark Scheduler内部原理剖析：http://sharkdtu.com/posts/spark-scheduler.html
Spark入门实战系列--4.Spark运行架构：https://blog.csdn.net/yirenboy/article/details/47441465
Spark调度（一）：Task调度算法，FIFO还是FAIR：
https://ieevee.com/tech/2016/07/11/spark-scheduler.html
《Spark 官方文档》Spark作业调度：
http://ifeve.com/spark-schedule/
yarn 调度器Scheduler详解：
https://blog.csdn.net/suifeng3051/article/details/49508261
-
spark checkpoint 的作用？
checkpoint的意思就是建立检查点,类似于快照,例如在spark计算里面 计算流程DAG特别长,服务器需要将整个DAG计算完成得出结果,但是如果在这很长的计算流程中突然中间算出的数据丢失了,spark又会根据RDD的依赖关系从头到尾计算一遍,这样子就很费性能,当然我们可以将中间的计算结果通过cache或者persist放到内存或者磁盘中,但是这样也不能保证数据完全不会丢失,存储的这个内存出问题了或者磁盘坏了,也会导致spark从头再根据RDD计算一遍,所以就有了checkpoint,其中checkpoint的作用就是将DAG中比较重要的中间数据做一个检查点将结果存储到一个高可用的地方(通常这个地方就是HDFS里面)。
特别Spark Streaming 会 checkpoint 两种类型的数据：
- Metadata（元数据） checkpointing - 保存定义了 Streaming 计算逻辑至类似 HDFS 的支持容错的存储系统。用来恢复 driver，元数据包括：
- 配置 - 用于创建该 streaming application 的所有配置
- DStream 操作 - DStream 一些列的操作
- 未完成的 batches - 那些提交了 job 但尚未执行或未完成的 batches
- Data checkpointing - 保存已生成的RDDs至可靠的存储。这在某些 stateful 转换中是需要的，在这种转换中，生成 RDD 需要依赖前面的 batches，会导致依赖链随着时间而变长。为了避免这种没有尽头的变长，要定期将中间生成的 RDDs 保存到可靠存储来切断依赖链
Checkpoint原理机制:
- 通过调用SparkContext.setCheckpointDir方法来指定进行Checkpoint操作的RDD把数据放在哪里，在生产集群中是放在HDFS上的，同时为了提高效率在进行checkpoint的使用可以指定很多目录
- 在进行RDD的checkpoint的时候其所依赖的所有的RDD都会从计算链条中清空掉；
- 作为最佳实践，一般在进行checkpoint方法调用前通过都要进行persist来把当前RDD的数据持久化到内存或者磁盘上，这是因为checkpoint是Lazy级别，必须有Job的执行且在Job执行完成后才会从后往前回溯哪个RDD进行了Checkpoint标记，然后对该标记了要进行Checkpoint的RDD新启动一个Job执行具体的Checkpoint的过程；
- Checkpoint改变了RDD的Lineage；
- 当我们调用了checkpoint方法要对RDD进行Checkpoint操作的话，此时框架会自动生成RDDCheckpointData，当RDD上运行过一个Job后就会立即触发RDDCheckpointData中的checkpoint方法，在其内部会调用doCheckpoint，实际上在生产环境下会调用ReliableRDDCheckpointData的doCheckpoint，在生产环境下会导致ReliableCheckpointRDD的writeRDDToCheckpointDirectory的调用，而在writeRDDToCheckpointDirectory方法内部会触发runJob来执行把当前的RDD中的数据写到Checkpoint的目录中，同时会产生ReliableCheckpointRDD实例；
具体使用：
sc.setCheckpointDir("hdfs://lijie:9000/checkpoint0727")
val rdd = sc.parallelize(1 to 10000)
rdd.cache()
rdd.checkpoint()
因为执行checkpoint的时候会重新从头开始计算partition数据，所以一般我们先进行cache然后做checkpoint就会只走一次流程,checkpoint的时候就会从刚cache到内存中取数据写入hdfs中。
checkpoint 的形式:
最终 checkpoint 的形式是将类 Checkpoint的实例序列化后写入外部存储，值得一提的是，有专门的一条线程来做将序列化后的 checkpoint 写入外部存储。类 Checkpoint 包含以下数据
Checkpoint类
除了 Checkpoint 类，还有 CheckpointWriter 类用来导出 checkpoint，CheckpointReader 用来导入 checkpoint
参见：
Spark中的checkpoint作用与用法：https://blog.csdn.net/qq_20641565/article/details/76223002
【容错篇】Spark Streaming的还原药水——Checkpoint：
//www.greatytc.com/p/00b591c5f623
Spark CheckPoint彻底解密（41）：http://blog.51cto.com/5233240/1773649
-
spark实际操作题1：
输入文件格式如下：
person ，friends....
100, 200 300 600 700
200, 100 500 700
300, 100 600 800 900
400, 600 900
500, 200
要求通过spark输出得到两个人之间的共同好友（注意：a是b好友 <=> b是a好友）。
这道自己想到的思路有三种，
- 第一种是在网上看到的，如下func1所示：
def func1() {
val sc = SparkSession.builder()
.appName("spark实际操作题1[RDD]")
.master("local[2]")
.getOrCreate()
.sparkContext
// sc.setCheckpointDir("F:\\storage\\dafo01\\E\\workSpace\\idea_project\\spark_test\\src\\main\\scala\\some\\spark实际操作题1[RDD]_checkpoint")
val textRdd = sc.textFile("F:\\storage\\dafo01\\E\\workSpace\\idea_project\\spark_test\\src\\main\\scala\\some\\spark实际操作题1")
.cache()
// textRdd.checkpoint()
val pairs = textRdd.flatMap(line => {
val tokens = line.split(",")
val person = tokens(0).toLong
val friends = tokens(1).split("\\s+").map(_.toLong).toList
val result = for {
i <- 0 until friends.size
friend = friends(i)
} yield {
if (person < friend)
((person, friend), friends)
else
((friend, person), friends)
}
// val result = ListBuffer.empty[((Long, Long), List[Long])]
// for(friend <- friends)
// {
// if (person < friend)
// result += new Tuple2(new Tuple2(person, friend), friends)
// else if (person > friend)
// result += new Tuple2(new Tuple2(person, friend), friends)
// }
result
})
val grouped = pairs.groupByKey()
val commonFriends = grouped.mapValues(iter => {
val friendCount = for {
list <- iter
if !list.isEmpty
friend <- list
} yield ((friend, 1))
// val friendCount = ListBuffer.empty[(Long, Int)]
// for (list <- iter) {
// if (!list.isEmpty) {
// for (friend <- list) {
// friendCount.+=(Tuple2(friend, 1))
// }
// }
// }
friendCount.groupBy(_._1).mapValues(_.unzip._2.sum).filter(_._2 > 1).map(_._1)
})
val formatedResult = commonFriends.map(
f => s"(${f._1._1},${f._1._2})\t${f._2.mkString("[", "，", "]")}"
)
formatedResult.foreach(println)
sc.stop()
}
}
主要思路是先找到好友对（p1,p2），再分别针对p1，p2绑定其好友成((p1，p2),friends_of_p1）和((p1,p2),friends_of_p2)，然后找到friends_of_p1和friends_of_p2之间的共同数据项，其中friends_of_p1位p1所对应好友list。
可以看到通过func1的一顿操作就能实现相应功能，但自己感觉实现略有点复杂，中间有大量的list_of_friends浪费内存，而且其实现只对p1，p2互为好友的情况有用，若p1，p2有共同好友但彼此互不为好友，则无法操作。
- 第二种方法是自己看到网上这种方法繁琐，结合数据想到的一种解法，如下func2：
def func2: Unit = {
val sc = SparkSession.builder()
.appName("spark实际操作题1[RDD]")
.master("local[2]")
.getOrCreate()
.sparkContext
sc.textFile("F:\\storage\\dafo01\\E\\workSpace\\idea_project\\spark_test\\src\\main\\scala\\some\\spark实际操作题1").cache()
.flatMap(line => {
val tokens = line.split(",")
val people = tokens(0).toLong
val friends = tokens(1).split("\\s+").map(_.toLong).filter(_ != people).toSeq
val friends_pairs = for (f1 <- friends; f2 <- friends; if f1 < f2)
yield (f1, f2)
for (friends_pair <- friends_pairs)
yield (friends_pair, people)
})
.groupByKey()
.map(x => s"${x._1}: 共同好友数（${x._2.size}）,他们是${x._2.mkString("[", ",", "]")}")
.foreach(println)
sc.stop()
}
材料给出的数据是people和对应的该people的所有好友，我们反过来理解也就是说people对应的friends两两之间都以该people为共同好友，那么问题迎刃而解了。相比第一种，没有浪费存储空间，而且实现简单。
- 第三种思路，利用rdd转dataframe，利用sparksql来解决数据查找问题，如func3所示：
def func3: Unit = {
val spark = SparkSession.builder()
.appName("spark实际操作题1[RDD]")
.master("local[2]")
.getOrCreate()
import spark.implicits._
val friendsDF = spark.sparkContext
.textFile("F:\\storage\\dafo01\\E\\workSpace\\idea_project\\spark_test\\src\\main\\scala\\some\\spark实际操作题1")
.flatMap(line => {
val tokens = line.split(",")
tokens(1).split("\\s+")
.map(Friends(tokens(0), _))
})
.toDF()
friendsDF.createOrReplaceTempView("friends")
spark.sql("select a.from as A,b.from as B,collect_set(a.to) as common,count(*) as count" +
" from friends a " +
" join friends b on a.to = b.to and a.from < b.from " +
" group by A,B" +
" order by A,B")
.persist(StorageLevel.MEMORY_ONLY)
.show(Int.MaxValue)
spark.stop()
}
充分利用强大的spark_sql功能，活学活用，见招拆招。
-
对于来自订单系统和页面系统的突发大数据，怎样分区来应对？
大多数人面试的时候都在想应该根据业务的数据量来分配节点，怎么怎么怎么样。。。。其实可能被这两个系统的的条件给限制了，因为大多数企业都会将这些系统的数据全部打散混合到一起（而不是想怎么把数据分到不同的节点上），通过hash或者随机数的方式，这些数据很自然的就会均衡分布到对应的节点上，当这些数据到达终点的时候再将它们分开，分别处理。
-
spark streaming怎么处理数据倾斜？
一般sparkstreaming用于处理如PV（page view：浏览页面总和）、UV（unique visitor：访客数）和DAU（daily active user：日活跃用户），又spark streaming的逻辑其实是将数据流切片，然后按照spark的方法来处理数据，在网上找了一些spark处理数据倾斜的方法，大概有以下几种：
- 最简单的方法是直接通过调整（增大或者减小）并行度来解决数据倾斜，主要是针对好几个大数据量的key分配到一个partition的情况，治标不治本，好在简单，遇到问题的时候可以先看看这种方法是否可行。
- 自定义partitioner，调动groupby这类分区函数的时候可以指定实现定制的parttionner，这种方法主要是针对特定的业务或数据，在通用性上面不一定具有优势，但还比较好用吧~
- 针对jion操作两张表的时候，性能消耗主要集中reduce join 的 shuffle阶段的 IO ，如果一张表的数据不多，我们可以将小表的数据加载到driver中，并通过broadcast将其广播到各个excutor中去，当我们再次执行join的时候，就能在个节点本机处理join数据，跳过shuffle阶段，化reduce join 为 map join ，从而提高处理能力。
- 当有少数几个key数据量提别大的时候，而另一张数据表相对分布均匀的时候，一般可以将这几个大数据量key单独分离出来，通过在key前面加随机数的方式分散数据到不同的task，最后union 这几个key的数据和剩下key的数据，便能有效解决。
- 另一种情况是一张表中有很多大数据量的key，这样的话第4种方法就没那么好使，对于这种情况，我们可以试着为所有的key都加上随机数前缀，数据分散到不同的excutor之后就迎刃而解了。但要注意的是这种方法也是处理大小表的情况，而且小表最好扩大与随机数取值范围相同的倍数（存储资源消耗也相应变大了N倍），这样大小表都能在一个excutor中实现join操作，避免了多余的IO操作。
补充一些知识：
spark foreachpartition和foreach的区别：
foreach传递的是每条数据，而foreachpartition传递的是每个分区的迭代器iterator。foreach 相当于 foreachpartition再加一层foreach，所以foreachpartition 可以节约不少生成中间变量的资源。同样的道理也适用于mappartition。
-
以（key，value）形式组成的数组 a=[('key1',4),('key2',5),('key1',2),('key3',4),('key1',6)],求：（1）数组a内value的和（2）分别计算数组a下每个key下value之和（3）使用Accumulator统计数组内元素总个数
这题主要是考察一些基本知识，答案如下：
val conf = new SparkConf().setAppName("Simple Application").setMaster("local[2]")
val sc = new SparkContext(conf)
val a = Array(("key1", 4), ("key2", 5), ("key1", 2), ("key3", 4), ("key1", 6))
val rdd = sc.parallelize(seq=a,numSlices = 2)
//(1)
println(rdd.map(_._2).sum())
//(2)
rdd.reduceByKey(_+_).foreach(println)
//(3)
val acum = sc.longAccumulator
rdd.foreach( x =>acum.add(1))
println(acum)
这里需要注意一下第三题的accumulator，spark为集群中不同节点共享数据提供了两个受限的共享变量，广播变量和累加器。其中广播变量需要在driver设置，broadcast到各个executor后就变成只读变量了。而累加器是driver中设置，并只能在driver中读取的变量，当被分发到各个节点之后各节点不能读取但可通过关联操作进行加运算，常用来做计数和求和。
-
有一名字为sc-data-1.log的日志文件位于Hadoop集群的/tmp/test,格式如下：[时间][日志类别][MainThread:ID][程序名字:行号][] - 消息，问题如下：（1）配置spark启用2个worker，每个worker节点分配5G内存,3个core.（2）请将[ERROR]日志类别中的程序名字以及行数输出到sc-data-1-reaults.txt并存入hdfs中。
第（1）题主要是考察你对spark基本配置的了解，为实现上述目标对conf目录下的spark-env.sh做如下修改：
SPARK_EXECUTOR_INSTANCES=2
SPARK_EXECUTOR_CORES=3
SPARK_EXECUTOR_MEMORY=5000M
第（2）题也是一些基础知识，代码如下：
val conf = new SparkConf().setAppName("Simple Application").setMaster("local[2]")
val sc = new SparkContext(conf)
val rdd = sc.textFile("hdfs://192.168.31.242:9000/tmp/test/sc-data-1.log")
val save_file = rdd.filter(_.contains("[INFO]"))
.map(line => {
val seq = line.split("\\[|\\]")
if (seq(3) == "INFO") //注意被切分后的数据位置
seq(7)
})
.saveAsTextFile("hdfs://192.168.31.242:9000/tmp/test/sc-data-1-result.txt")
同样需要注意的是：
1.split之后因为是匹配‘[’或‘]’切分，所以当遇到“][”的情况的时候，程序会将“][”之间的空白也算作一段字符串（虽然为空），这样就会影响到对于切分后数组下标。
2.是最后saveAsTextFile的时候，spark默认会将sc-data-1-result.txt保存为文件夹，如下：
其中.crc都为循环校验文件，_SUCCESS文件用于表明任务成功，part-xxxxx表明任务由不同的task完成（也对应相应的分区数），有多少task就有多少个part-xxxx文件，这也体现了spark任务分布式处理的特点。当然最后你也可以结合spark提供的collect（）方法将数据合并，再调用hadoop的hdfs组件，单独从driver上传合并后的文件到服务器，这样系统最后得到的就是一个文件，但会涉及更多IO开销，作为大数据系统，个人认为这样不是很符合分布式的思想，也没什么必要。
TO BE CONTINUED ......
