Spark 是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。最开始，Spark创建的目的就是解决Hadoop计算时间过长问题。
java 语言编写的，在分布式服务器集群上存储海量数据并运行分布式分析应用的开源框架HDFS理论基础源于Google的 TheGoogleFileSystem 论文，它是 GFS 的开源实现。MapReduce 论文实现。性能横向扩展变得非常容易。Bigtable 的开源实现，但又和 Bigtable 存在许多不同之处。HBase 是一个基于 HDFS 的分布式数据库，擅长实时地随机读/写超大规模数据集。它也是 Hadoop 非常重要的组件。Scala 语言开发的快速、通用、可扩展的大数据分析引擎。Spark Core 中提供了 Spark 最基础与最核心的功能。Spark SQL 是 Spark 用来操作结构化数据的组件。通过 Spark SQL，用户可以使用SQL 或者 Apache Hive 版本的 SQL 方言（HQL）来查询数据。Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的 API。Spark MLlib是 Spark 提供的一个机器学习算法库。MLlib 不仅提供了模型评估、数据导入等额外的功能，还提供了一些更底层的机器学习原语。Spark GraphX 是 Spark 面向图计算提供的框架与算法库。Spark 支持 DAG，能够满足 并行复用 场景
Hadoop MapReduce 由于其设计初衷并不是为了满足循环迭代式数据流处理，因此在多并行可复用场景中存在诸多计算效率等问题。所以 Spark 应运而生(支持DAG, 数据复用性强)，Spark 就是在传统的 MapReduce 计算框架的基础上，利用其计算过程的优化，从而大大加快了数据分析、挖掘的运行和读写速度，并将计算单元缩小到更适合并行计算和重复使用的 RDD 计算模型。
Spark 和Hadoop 的根本差异是多个作业之间的数据通信问题 : Spark 多个作业之间数据通信是基于内存，而 Hadoop 是基于磁盘。
Spark Task 的启动时间快。Spark 采用 fork 线程的方式，而 Hadoop 采用创建新进程(通过JVM重用进行优化)的方式。
Spark 只有在 shuffle 的时候将数据写入磁盘，而 Hadoop 中多个 MR 作业之间的数据交互都要依赖于磁盘交互
Spark 支持 checkpoint，保存数据切断血缘，重用性大大增强。
经过上面的比较，我们可以看出在绝大多数的数据计算场景中，Spark 确实会比 MapReduce更有优势。但是 Spark 是基于内存的，所以在实际的生产环境中，由于内存的限制，可能会由于内存资源不够导致 Job 执行失败，此时，MapReduce 其实是一个更好的选择，所以 Spark并不能完全替代 MR。
Spark 框架的核心是一个计算引擎，整体来说，它采用了标准 master-slave 的结构。Driver 表示 master，负责管理整个集群中的作业任务调度。图形中的 Executor 则是 slave，负责实际执行任务。
Driver
Spark 驱动器节点，用于执行 Spark 任务中的 main()，负责实际代码的执行工作。Driver 在 Spark 作业执行时主要负责：
job）task)Executor
Spark Executor 是集群中工作节点（Worker）中的一个 JVM 进程，负责在 Spark 作业中运行具体任务（Task），任务彼此之间相互独。Spark 应用启动时，Executor 节点被同时启动，并且始终伴随着整个 Spark 应用的生命周期而存在。如果有 Executor 节点发生了故障或崩溃，Spark 应用也可以继续执行，会将出错节点上的任务调度到其他 Executor 节点上继续运行(同MapReduce)。核心功能：
块管理器(Block Manager)为用户程序中要求缓存的 RDD 提供内存式存储。RDD 是直接缓存在 Executor 进程内的，因此任务可以在运行时充分利用缓存数据加速运算。Spark 集群的独立部署环境中，不需要依赖其他的资源调度框架，自身就实现了资源调度的功能，所以环境中还有其他两个核心组件：Master 和 Worker，这里的 Master 是一个进程，主要负责资源的调度和分配，并进行集群的监控等职责(相当于 Yarn 的 ResourceManager)。而Worker 呢，也是进程，一个 Worker 运行在集群中的一台服务器上，由 Master 分配资源对数据进行并行的处理和计算(类似于 Yarn 环境中 NodeManager)。
缺点：只支持简单的固定资源分配策略，每个任务固定数量的 core，各 Job 按顺序依次分配在资源，资源不够的时候就排队，一般使用 Yarn 实现资源调度。
Hadoop 用户向 YARN 集群提交应用程序时，提交程序中应该包含 ApplicationMaster(相当于 Yarn的MRAppMaster)，用于向资源调度器申请执行任务的资源容器 Container，运行用户自己的程序任务 job，监控整个任务的执行，跟踪整个任务的状态，处理任务失败等异常情况。说的简单点就是，ResourceManager（资源）和 Driver（计算）之间的解耦合靠的就是ApplicationMaster。
RDD : 弹性分布式数据集
是 Spark 中最基本的数据处理模型。是一个抽象类，它代表一个弹性的、不可变(产生新的)、不保存数据、可分区、里面的元素可并行计算的集合。主要用于将逻辑进行封装，并生成 Task 发送给Executor 节点执行计算
ACC 累加器：分布式共享只写变量
累加器用来把 Executor 端变量信息聚合到 Driver 端。在 Driver 程序中定义的变量，在Executor 端的每个 Task 都会得到这个变量的一份新的副本，每个 task 更新这些副本的值后，传回 Driver 端进行 merge，从而实现共享可写功能。
BC 广播变量：分布式共享只读变量
广播变量用来Driver向Executor高效分发一个较大的只读值，以供一个或多个 Spark 操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表。在多个并行操作中使用同一个变量，Spark 会为每个任务分别发送。实现共享可读功能。
广播变量在Driver定义与赋值，Executor端只读。Driver无法将一个RDD广播出去，因为RDD无法存储数据，但是可以将RDD结构广播出去，如 rdd.collect( )。广播变量一般存储在 Execution 区以实现共享。
Transformation算子：即从现有的数据集RDD创建一个新的数据集RDD，所以叫做 Transformation 转化算子。
Action算子：即在数据集上进行计算后，返回一个值给 Driver 程序。
RDD 中所有的 Transformation 都是惰性的，也就是说，它们并不会直接计算结果。相反的它们只是记住了这些 应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给 Driver 的 Action 时，这些 Transformation 才会真正运行，这个设计让 Spark 更加有效的运行。
RDD 根据数据处理方式的不同将算子整体上分为 Value 类型、双 Value 类型和 Key-Value 类型
Value 类型
groupBy：将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为 shuffle。极限情况下，数据可能被分在同一个分区中。可能会出现数据倾斜。coalesce：根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少分区的个数，减小任务调度成本。可能发生shuffle，参数控制。
repartition：该操作内部其实执行的是 coalesce 操作，参数 shuffle 的默认值为 true。无论是将分区数多的RDD 转换为分区数少的 RDD，还是将分区数少的 RDD 转换为分区数多的 RDD，repartition操作都可以完成，因为无论如何都会经 shuffle 过程。sortBy：该操作用于排序数据。在排序之前，可以将数据通过 f 函数进行处理，之后按照 f 函数处理的结果进行排序，默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程。双 Value 类型
RDD中的元素，Value 为第 2 个 RDD 中的相同位置的元素。Key-Value 类型
partitionBy：将数据按照指定 Partitioner 进行重分区。Spark 默认的分区器是 HashPartitioner。
reduceByKey：可以将数据按照相同的 Key 对 Value 进行聚合，在shuffle之前有combine（预聚合）操作，但不是任何场景都适用。
groupByKey：将数据源的数据根据 key 对 value 进行分组。
aggregateByKey：将数据根据不同的规则进行分区内计算和分区间计算。
foldByKey：当分区内计算规则和分区间计算规则相同时，aggregateByKey 就可以简化为 foldByKey。
combineByKey：最通用的对 key-value 型 rdd 进行聚集操作的聚集函数（aggregation function）。类似于aggregate()，combineByKey()允许用户返回值的类型与输入不一致。
sortByKey：在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口(特质)，返回一个按照 key 进行排序。
join：在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的 (K,(V,W)) 的 RDD。
leftOutJoin：类似于 SQL 语句的左外连接。
cogroup：在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable,Iterable))类型的 RDD。
reduce
聚集 RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据
collect
在驱动程序中，以数组 Array 的形式返回数据集的所有元素
count
返回 RDD 中元素的个数
first
返回 RDD 中的第一个元素
take
返回一个由 RDD 的前 n 个元素组成的数组
takeOrdered
返回该 RDD 排序后的前 n 个元素组成的数组
aggregate
分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合
fold
折叠操作，aggregate 的简化版操作
countByKey
统计每种 key 的个数
save
将数据保存到不同格式的文件中
foreach
分布式遍历 RDD 中的每一个元素，调用指定函数
crontroller 即控制算子，不直接参与计算，但是对性能和效率的有很好的支持。
比如 cache 和 persist，详见 [Cache/Persist 缓存](#3.4.1. RDD Cache/Persist 缓存)
RDD 任务切分中间分为：Application、Job、Stage 和 Task
Application：初始化一个 SparkContext 即生成一个 Application；
Job：一个 Action 算子就会生成一个 Job( runJob ( ) )；
Stage：Stage 等于宽依赖(ShuffleDependency)的个数加 1(第一个创建的ResultStage阶段)；
Task：一个 Stage 阶段中，最后一个 RDD 的分区个数就是 Task 的个数，而Task数量其实就是并行度。
注意：Application->Job->Stage->Task 每一层都是 1 对 n 的关系。
用户通过不同的线程提交的Job可以并发运行，但是受到资源的限制。Job到调度池（pool）内申请资源，调 度池会根据工程的配置，决定采用哪种调度模式。
FIFO模式 (先进先出)
在默认情况下，Spark调度器以FIFO（先进先出）方式调度Job的执行。每个Job被切分为多个Stage。无论是Job还是Stage，都是前面的优先分配。如果前面的资源分配完成，还剩下资源的话，后面的Job可以获取剩下的，可能存在多个Job并行运行。
FAIR模式（公平）
从spark0.8开始，可以配置公平调度器，在FAIR共享模式调度下，Spark在多Job之间以轮询（round robin）方式为任务分配资源，所有的任务拥有大致相当的优先级来共享集群的资源。这就意味着当一个长任务正在执行时，短任务仍可以分配到资源，提交并执行， 并且获得不错的响应时间。这样就不用像以前一样需要等待长任务执行完才可以。这种调度模式很适合多用户的场景。
RDD task任务 的执行是分 stage阶段的，后一个阶段必须要等前一个阶段执行完成才能执行。
DAGSchdule，开始划分阶段 createResultStage。ResultStage。getOrCreateParentStages。ShuffleDependency。getShuffleDependencies，获取每一个Shuffle依赖的阶段，如果没有就新建一个阶段getOrCreateShuffleMapStage。RDD 只支持粗粒度转换，即在大量记录上执行的单个操作。将创建 RDD 的一系列 Lineage（血缘）记录下来，以便恢复丢失的分区。RDD 的 血缘 会记录 RDD 的元数据信息和转换行为，当该 RDD 的部分分区数据丢失时，它可以根据这些信息来以及checkpoint来重新运算和恢复丢失的数据分区。相邻的 RDD 存在 依赖关系，即子RDD依赖父RDD，多个连续的 RDD 依赖关系称之为 血缘关系。
窄依赖表示每一个父(上游)RDD 的 Partition 最多被子(下游)RDD 的一个 Partition 使用，窄依赖相当于独生子女，窄依赖联系的父RDD和子RDD可以共享一个Task。窄依赖可以支持在同一个集群节点上，执行多条命令（也叫同一个 stage 的操作）， 例如在执行了 map 后，紧接着执行 filter。而且窄依赖计算失败后只需要重算失败的那个分区数据即可。
宽依赖表示同一个父(上游)RDD 的 Partition 被多个子(下游)RDD 的 Partition 依赖，会引起 Shuffle 详见 [Shuffle机制](#3.7. Shuffle 机制)，宽依赖相当于 二胎三胎。由于宽依赖会产生Shuffle，需要两个Task分阶段分别完成父RDD和子RDD。宽依赖计算失败，需要将关系的所有分区数据都重算才可以。
当父RDD与子RDD形成宽依赖关系时，中间需要进行Shuffle。Shuffle的主要目的就是将父RDD的分区数据根据配置的分区规则(默认hashPartitoner)，划分为子RDD需要的数量和规则要求的分区数据。
Spark的每个 Stage 阶段都会有一次 Shuffle，根据 Stage 的 Shuffle 特点，将 Task 分为 ShuffleMapTask 和 ResultTask。其中 ResultTask 就是Stage最后一个也是经过 Shuffle 之后的 Task，其它都是 ShuffleMapTask。
由 ShuffleMapTask 到 ResultTask，主要有以下两个 Shuffle 策略：
Shuffle write
每一个ShufflleMapTask会为每一个ResultTask创建一个bucket缓存，并且会为每一个bucket创建一个文件block file。这个bucket存放的数据就是经过Partitioner操作(默认是HashPartitioner)之后找到对应的bucket然后放进去，最后将数据bucket缓存数据写入磁盘block flie。
Shuffle read
ResultTask 会讲需要的所有block file 读取到内存当中，如果内存不足，同时时候磁盘。
缺点是每个 ShuffleMapTask 都会产生 reduce 数量个 buket 缓存和 block 文件。如果 ShuffleMapTask 数量很多而且数据量很大，首先内存方面很容易不足，然后就是需要持久化的文件个数过多，效率不高。
优化后的 HashShuffle
优化内容：
有原来的每个Task输出block flie改编为每个Executor（Executor内部可能对应多个task）输出block file。比优化前的 hashshuffle 的文件数量少了很多。
优点：
同一个 Executor 的所有并行 ShuffleMapTask 可以将数据输出到同一个blockflie，大大减少 block file 数量，提高性能。
缺点：
普通的 SortShuffle
Spark 借鉴了 MapReduce 的 Shuffle 处理，使得 ShuffleMapTask 输出的数据文件数量等于 ResultTask 数量个，并且有序(相当于将Reduce阶段的计算放在Map阶段，实现优化)。步骤如下：
bypass的SortShuffle
触发条件：
优势：
HashPartitioner：
Spark 在做 Shuffle 时，默认使用 HashPartitioner对数据进行分区。如果并行度设置的不合适，可能造成大量不相同的 Key 应的数据 被分配到了同一个 Task 上，造成该 Task 所处理的数据远大于其它 Task，从而可能造成数据倾斜。如果调整 Shuffle 时的并行度，使得原本被分配到同一 Task 的不同 Key 发配到不同 Task 上处理，则可降低原 Task 所需处理的数据量，从而缓解数据倾斜问题造成的短板效应。
RangerPartitioner：
首先使用 水塘抽样 获取样品数据，再通过内部计算得到边界数组，将数据按照边界数据进行范围划分，以范围为单位划分给每个分区，分界的算法尤为重要，对应的函数是rangeBounds。 。数据比较均衡，并且分区内数据是个范围，实现分区间有序。
自定义分区器：
RDD 通过 Cache(本质是persist(StorageLevel.MEMORY_ONLY)) 或者 Persist 方法将前面的计算结果(触发action 算子时)缓存，该 RDD 将会被缓存在计算节点的内存中，并供后面多路重用或失败重算。
Spark 会自动对一些 Shuffle 操作的中间数据做持久化操作(比如：reduceByKey)。这样做的目的是为了当一个节点 Shuffle 失败了避免重新计算整个输入。但是，在实际使用的时候，如果想重用数据，仍然建议调用 persist 或 cache。
所谓的检查点其实就是通过将 RDD 中间结果写入磁盘。
由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重算血缘，减少了开销。对 RDD 进行 checkpoint 操作并不会马上被执行，必须执行 Action 操作才能触发。
切断血缘依赖。HDFS 等容错、高可用的文件系统，可靠性高。建议对CK的RDD 使用 Cache 缓存，这样 CK 的 job 只需从 Cache 缓存中读取数据即可，否则需要再从头计算一次 RDD。
Spark部署方式，详见[Spark部署方式](#4.2. Spark 部署方式)。实际选用 Yarn ClusterX方式。
脚本启动，SparkSubmit创建一个 Yarn Client，通过反射执行main( )。
spark-submit \
--master local[5] \
--driver-cores 2 \ # driver使用内核数，默认为1
--driver-memory 8g \ # driver内存大小，默认512M
--executor-cores 4 \ # 每个executor内核数，默认为1，官方建议2-5个，我们企业是4个
--num-executors 10 \ # 启动executors的数量，默认为2
--executor-memory 8g \ # executor内存大小，默认1G
--class PackageName.ClassName XXXX.jar \
--name "Spark Job Name" \
InputPath \
OutputPath
DAGSchedule ：
TaskSchedule：
Local：运行在一台机器上，通常是练手或者测试环境。Standalone：构建一个基于Master+Slaves的资源调度集群，Spark任务提交给Master运行。是Spark自身的一个调度系统。 缺点：只支持简单的固定资源分配策略，每个任务固定数量的 core，各 Job 按顺序依次分配在资源，资源不够的时候就排队。Yarn Client：Client 模式将用于监控和调度的 Driver 模块在Yarn 客户端执行，而不是在 Yarn 中，所以一般用于测试。优点是方便调试及查看各种日志，缺点是Client连接中断，任务就会失败。Yarn Cluster：Cluster 模式将用于监控和调度的 Driver 模块启动在 Yarn 集群资源中执行。一般应用于实际生产环境。优点是Client连接中断，任务继续执行，日志需要登录某个节点查看。Yarn 模式的优点：
DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。
Spark 将计算程序根据依赖关系映射成点与线组成的 DAG 有向无环图，并且划分层级，详见 [层级划分](#3.3.5. RDD作业的划分)。
Job 和 Stage 存在执行顺序，串行处理，而 多个Task 之前并没有依赖关系，可以并行处理。关于 Stage 的划分，其实是根据RDD依赖关系划分的，详细见 [RDD依赖关系](#3.3. RDD依赖关系)。
Spark将一个Application划分为多个Job作业，每个Job作业又划分多个Stage，而Executor的Task线程实际执行的是Stage的 taskset。Spark的并行指的某个Stage阶段，同时有多个Task线程在执行task任务。
Spark的并行的Task，一个Task处理一个分区的数据。如果要确定下游的并行结构，就需要确定两个因素。就是 并行度(即分区数) 和 分区规则。简单来说就是先确定有多少的并行Task线程，再将各个并行Task线程的输入数据按照分区规则分配给执行的Task线程。
每个Stage的并行度，代表了这个Stage的同时计算能力，并行度越大，处理相同数据量时间越小，并行度的影响因素主要有 资源并行度 和 数据并行度。有足够的资源才能启动足够数量的并行Task线程，有了足够的并行Task线程，就可以为每一个分区数据进行并行计算。
资源的并行度:
由节点数( executor )和 cpu 数( core )决定的。
–num-executors(Executor 的数量)；
–executor-memory(每个 Executor 的内存大小)；
–executor-cores( Executor 的 core 数量)。
数据的并行度: 并行 task 的数据，parttion 大小。
影响分区的因素：
启动时置顶默认并行度 spark.default.parallelism ， 或 SparkContext 的 setMaster("local[*]")。
对于Scala集合分发到RDD，可以使用 sc.parallelize(scala集合, 并行度)。优先级高于spark.default.parallelism，但是不要大于这个值。
val rdd: RDD[Int] = sc.parallelize(List(1, 2, 3), 3)
* 读取文件是设置并行度
1. `textFile(path，minPartitions)`
```scala
// partition最大值是2
math.min(defaultParallelism, 2);
```
2. `wholeTextFile(path, minPartitions)`
```scala
// partition最小值是2
max(minPartitions, 2);
```
3. Shuffle 改变分区数量。比如 rePartition，groupBykey...
4. 自定义分区改变分区数量。
如何合理设置 资源 及 并行度(task数量)：
Task数量 与 资源关系
- task数量少，资源剩余，直接浪费了差值资源。
- task数量和资源数量相同，先完成的task资源无法及时释放。
- task数量多，资源不足，执行完一批task，在执行下一批task，资源重用。
所以，task数量要相应多于资源数量，官方推荐
Task数量为 core 总数 2～3 倍最佳。在
合理配置初始并行度和资源时候，先再根据数据量，确定初始Task数量，反推资源数量(2～3倍)，完成合理配置。
并行度的确定，只是确定了下游有几个Task线程，但是无法知道每一个Task线程需要计算的数据是什么。Spark是 一个Task线程处理一个分区的数据，而上游分区分配到下游分区，需要分区器来定义分区规则。
对 Spark/Hadoop 这样的大数据系统来讲，数据量大并不可怕，可怕的是数据倾斜。数据倾斜指的是，并行处理的数据集中，某一部分（如 Spark 或 Kafka 的一个 Partition）的数据显著多于其它部分，从而使得该部分的处理速度成为整个数据集处理的瓶颈（木桶效应）。
在 Spark 中，同一个 Stage 的不同 Partition 可以并行处理，而具有依赖关系的不同 Stage 之间是串行处理的。数据倾斜往往发生在同一个Stage不同的task中，即个别task计算时间远远长于其他task。而相同 Stage 内的所有 Task 都执行相同的计算，在排除不同计算节点计算能力差异的前提下，不同 Task 之间耗时的差异主要由该 Task 所处理的数据量决定。
Spark 数据倾斜的几种场景及对应的解决方案 调整并行度，自定义Partitioner，使用 Map侧Join 代替 Reduce侧Join（内存表合并），给倾斜 Key 加上随机前缀等。
调整并行度分散同一个 Task 的不同 Key
HashPartitioner：
Spark 在做 Shuffle 时，默认使用 HashPartitioner对数据进行分区。如果并行度设置的不合适，可能造成大量不相同的 Key 应的数据 被分配到了同一个 Task 上，造成该 Task 所处理的数据远大于其它 Task，从而造成数据倾斜。如果调整 Shuffle 时的并行度，使得原本被分配到同一 Task 的不同 Key 发配到不同 Task 上处理，则可降低原 Task 所需处理的数据量，从而缓解数据倾斜问题造成的短板效应。
RangerPartitioner：
首先使用 水塘抽样 获取样品数据，再通过内部计算得到边界数组，将数据按照边界数据进行范围划分，以范围为单位划分给每个分区。数据比较均衡，并且分区内数据是个范围。
自定义Partition
使用自定义的 Partitioner（默认为 HashPartitioner），将原本被分配到同一个 Task 的不同Key 分配到不同 Task，可以拿上图继续想象一下，通过自定义 Partitioner 可以把原本分到 Task0 的 Key分到Task1，那么 Task0 的要处理的数据量就少了。
将 Reduce side Join 转变为 Map side Join
通过 Spark 的 Broadcast 机制，将 Reduce side Join 转化为 Map side Join，避免 Shuffle 从而完全消除 Shuffle 带来的数据倾斜，但是有内存溢出的风险。
为 skew 的 key 增加随机前/后缀
为数据量特别大的 Key 增加随机前/后缀，使得原来 Key 相同的数据变为Key 不相同的数据，从而使倾斜的数据集分散到不同的 Task 中，彻底解决数据倾斜问题。Join 另一则的数据中， 倾斜 Key 对应的部分数据，与随机前缀集作笛卡尔乘积，从而保证无论数据倾斜侧的 Key 如何加前缀，都能正常 Join。
大表随机添加 N 种随机前缀，小表扩大 N 倍
如果出现数据倾斜的 Key 比较多，上一种方法将这些大量的倾斜Key 分拆出来，意义不大（很难一个 Key 一个 Key 都加上后缀）。此时更适合直接对存在数据倾斜的数据集全部 加上随机前缀，然后对另外一个不存在严重数据倾斜的数据集整体与随机前缀集作笛卡尔乘积（即将数据量扩大 N 倍），可以看到RDD2 扩大了 N 倍了，再和加完前缀的大数据做笛卡尔积。
Spark的Executor的内存管理是基于JVM的内存管理之上，Spark对JVM堆内(On-Heap)空间进行了更为详细的分配， 以便充分利用内存，同时Spark引入堆外内存(OffHeap)，可以直接在Worker节点的系统内存中开辟空间，进一步优化内存使用。
堆内空间(On-Heap)
Spark的堆内(On-Heap)空间是由--executor-memory参数配置，Executor内运行的并发任务共享JVM堆内内存。而且该堆内内存是一种逻辑上的管理，因为对象的释放都是由JVM完成。
堆外内存(Off-Heap)
主要是为了提高Shuffle排序的效率，存储优化过的二进制数据。从2.0之后Spark 可以直接操作系统的堆外内存，减少不必要的开销。默认不开启，spark.memory.offHeap.ennable 参数启用，并由spark.memory.offHeap.size 参数设定堆外空间大小。
Spark 1.6之前使用的是静态内存管理( StaticMemoryManager )机制 StaticMemoryManager 也是 Spark 1.6之前唯一的内存管理器。在Spark1.6之后引入了统一内存管理( UnifedMemoryManager )机制， 是 Spark 1.6之后默认的内存管理器。
默认情况下，Spark 仅仅使用了堆内内存。Executor 端的堆内内存区域大致可以分为以下四大块：
Execution 内存：主要用于存放 Shuffle、Join、Sort、Aggregation 等计算过程中的临时数据。Excutor内运行的Task，共享 Execution 内存。
Storage 内存：主要用于存储 spark 的 cache 数据，例如RDD的缓存、unroll数据。
Execution 和 Storage 内存中间存在内存动态调整机制，即任何一方内存不够而另一方内存空闲的情况下，可以申请到对方的一部分内存自己使用。当Execution 需要使用内存对方占用时候，可让对方将占用的部分转存到硬盘，然后"归还"借用的空间。反之 Storage 无法让对方 “归还”，这也是应该 Shuffle 存储情况复杂，可能会被后面任务使用。
User 内存：主要用于存储 RDD 转换操作所需要的数据，例如 RDD 依赖等信息。
预留内存：系统预留内存，会用来存储Spark内部对象。
关系：
两者都是用来改变RDD的partition数量的，repartition底层调用的就是coalesce方法：coalesce(numPartitions, shuffle = true)
区别：
repartition一定会发生shuffle，coalesce根据传入的参数来判断是否发生 shuffle。一般情况下增大rdd的partition数量使用repartition，减少partition数量时使用coalesce。
读取的数据源为非常多的小文件时，如果使用 textFile() 读取数据，那么一个文件就是一个 Task，并且要为每个文件建立一个分区，效率低下。这时候可以使用 wholeTextFiles进行大量小文件数据的读取，wholeTextFiles 通过设置的文件路径将小文件都读取完成后，会通过设置的 defaultParallelism，创建相应的分区和Task，如果没有设置，默认为2个。
场景描述：
I/O在很短的时间就可以完成，而 CPU 还有许多运算 要处理，CPU Loading 很高。一般是需要进行大量计算，比如逻辑判断，机器学习 或 图计算等，这是 CPU 成为性能瓶颈。
优化点：