目录

• 开发调优原则
  • 避免创建重复的RDD
  • 尽可能复用同一个RDD
  • 对多次使用的RDD持久化
    • Spark的持久化级别
    • 如何选择一种最合适的持久化策略
  • 尽量避免使用Shuffle类算子
    • Broadcast与map进行join代码示例
    • 使用Map-Side预聚合的Shuffle操作
  • 使用高性能的算子
    • 使用reduceByKey/aggregateByKey替代groupByKey
    • 使用mapPartitions替代普通map
    • 使用foreachPartitions替代foreach
    • 使用filter之后进行coalesce操作
    • 使用repartitionAndSortWithinPartitions替代repartition与sort类操作
  • 广播大变量
  • 使用Kryo优化序列化性能
    • 在 Spark 中，主要有三个地方涉及到了序列化：
  • 优化数据结构

开发调优原则

避免创建重复的RDD

对于同一份数据，只应该创建一个 RDD，不能创建多个 RDD 来代表同一份数据。否则 我们的 Spark 作业会进行多次重复计算来创建多个代表相同数据
的 RDD，进而增加了作业的性能开销。

// 需要对名为“hello.txt”的HDFS文件进行一次map操作，再进行一次reduce操作。也就是说，需要对一
份数据执行两次算子操作。
// 错误的做法：对于同一份数据执行多次算子操作时，创建多个RDD。
// 这里执行了两次textFile方法，针对同一个HDFS文件，创建了两个RDD出来，然后分别对每个RDD都执
行了一个算子操作。
// 这种情况下，Spark需要从HDFS上两次加载hello.txt文件的内容，并创建两个单独的RDD；第二次加载
HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的。
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
rdd1.map(...)
val rdd2 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
rdd2.reduce(...)
// 正确的用法：对于一份数据执行多次算子操作时，只使用一个RDD。
// 这种写法很明显比上一种写法要好多了，因为我们对于同一份数据只创建了一个RDD，然后对这一个RDD执
行了多次算子操作。
// 但是要注意到这里为止优化还没有结束，由于rdd1被执行了两次算子操作，第二次执行reduce操作的时
候，还会再次从源头处重新计算一次rdd1的数据，因此还是会有重复计算的性能开销。
// 要彻底解决这个问题，必须结合“原则三：对多次使用的RDD进行持久化”，才能保证一个RDD被多次使用
时只被计算一次。
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
rdd1.map(...)
rdd1.reduce(...)

尽可能复用同一个RDD

除了要避免在开发过程中对一份完全相同的数据创建多个 RDD 之外，在对不同的数据执行算子操作时
还要尽可能地复用一个 RDD。比如说，有一个 RDD 的数据格式是 key-value 类型的，另一个是单
value 类型的，这两个 RDD 的 value 数据是完全一样的。那么此时我们可以只使用 key-value 类型的那
个 RDD，因为其中已经包含了另一个的数据。对于类似这种多个 RDD 的数据有重叠或者包含的情况，
我们应该尽量复用一个 RDD，这样可以尽可能地减少 RDD 的数量，从而尽可能减少算子执行的次数。

// 错误的做法。
// 有一个<Long, String>格式的RDD，即rdd1。
// 接着由于业务需要，对rdd1执行了一个map操作，创建了一个rdd2，而rdd2中的数据仅仅是rdd1中的
value值而已，也就是说，rdd2是rdd1的子集。
JavaPairRDD<Long, String> rdd1 = ...
JavaRDD<String> rdd2 = rdd1.map(...)
// 分别对rdd1和rdd2执行了不同的算子操作。
rdd1.reduceByKey(...)
rdd2.map(...)
// 正确的做法。
// 上面这个case中，其实rdd1和rdd2的区别无非就是数据格式不同而已，rdd2的数据完全就是rdd1的子
集而已，却创建了两个rdd，并对两个rdd都执行了一次算子操作。
// 此时会因为对rdd1执行map算子来创建rdd2，而多执行一次算子操作，进而增加性能开销。
// 其实在这种情况下完全可以复用同一个RDD。
// 我们可以使用rdd1，既做reduceByKey操作，也做map操作。
// 在进行第二个map操作时，只使用每个数据的tuple._2，也就是rdd1中的value值，即可。
JavaPairRDD<Long, String> rdd1 = ...
rdd1.reduceByKey(...)
rdd1.map(tuple._2...)
**// 第二种方式相较于第一种方式而言，很明显减少了一次rdd2的计算开销。**
// 但是到这里为止，优化还没有结束，对rdd1我们还是执行了两次算子操作，rdd1实际上还是会被计算两
次。
// 因此还需要配合“原则三：对多次使用的RDD进行持久化”进行使用，才能保证一个RDD被多次使用时只被
计算一次。

对多次使用的RDD持久化

对多次使用的 RDD 进行持久化。此时 Spark 就会根据你的持久化
策略，将RDD 中的数据保存到内存或者磁盘中。以后每次对这个 RDD 进行算子操作时，都会直接从内
存或磁盘中提取持久化的 RDD 数据，然后执行算子，而不会从源头处重新计算一遍这个 RDD，再执行
算子操作。

// 如果要对一个RDD进行持久化，只要对这个RDD调用cache()和persist()即可。
// 正确的做法。
// cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。
// 此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。
// 第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。
val rdd1 = sc.textFile("hdfs://hadoop277ha/hello.txt").cache()
rdd1.map(...)
rdd1.reduce(...)
// persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化。
// 比如说，StorageLevel.MEMORY_AND_DISK_SER表示，内存充足时优先持久化到内存中，内存不充足
时持久化到磁盘文件中。
// 而且其中的_SER后缀表示，使用序列化的方式来保存RDD数据，此时RDD中的每个partition都会序列化
成一个大的字节数组，然后再持久化到内存或磁盘中。
// 序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，从而发
生频繁GC。
val rdd1 =
sc.textFile("hdfs://hadoop277ha/hello.txt").persist(StorageLevel.MEMORY_AND_DISK
_SER)
rdd1.map(...)
rdd1.reduce(...)

以上三点总结：
1、如果多个RDD的数据相同，记住，只需要创建，这一个即可，其他的地方，需要使用，复用这个RDD即可、
2、既然都复用了这个RDD，就应该把这个RDD进行持久化 (内存，磁盘)
cache() persist()

Spark的持久化级别

持久化级	含义解释
MEMORY_ONLY	使用未序列化的Java对象格式，将数据保存在内存中。如果内存不够存放所有的数据，则数据可能就不会进行持久化。那么下次对这个RDD执行算子操作时，那些没有被持久化的数据，需要从源头处重新计算一遍。这是默认的持久化策略，使用cache()方法时，实际就是使用的这种持久化策略。
MEMORY_AND_DISK	使用未序列化的Java对象格式，优先尝试将数据保存在内存中。如果内存不够存放所有的数据，会将数据写入磁盘文件中，下次对这个RDD执行算子时，持久化在磁盘文件中的数据会被读取出来使用。
MEMORY_ONLY_SER	基本含义同MEMORY_ONLY。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。
MEMORY_AND_DISK_SER	基本含义同MEMORY_AND_DISK。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。
DISK_ONLY	使用未序列化的Java对象格式，将数据全部写入磁盘文件中。
MEMORY_ONLY_2, MEMORY_AND_DISK_2, 等等.	对于上述任意一种持久化策略，如果加上后缀_2，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上。这种基于副本的持久化机制主要用于进行容错。假如某个节点挂掉，节点的内存或磁盘中的持久化数据丢失了，那么后续对RDD计算时还可以使用该数据在其他节点上的副本。如果没有副本的话，就只能将这些数据从源头处重新计算一遍了。

如何选择一种最合适的持久化策略

1、默认情况下，性能最高的当然是MEMORY_ONLY，但前提是你的内存必须足够足够大，可以绰绰有余地存放
下整个RDD的所有数据。因为不进行序列化与反序列化操作，就避免了这部分的性能开销；对这个RDD的后续
算子操作，都是基于纯内存中的数据的操作，不需要从磁盘文件中读取数据，性能也很高；而且不需要复制一
份数据副本，并远程传送到其他节点上。但是这里必须要注意的是，在实际的生产环境中，恐怕能够直接用这
种策略的场景还是有限的，如果RDD中数据比较多时（比如几十亿），直接用这种持久化级别，会导致JVM的
OOM内存溢出异常。
2、如果使用MEMORY_ONLY级别时发生了内存溢出，那么建议尝试使用MEMORY_ONLY_SER级别。该级别会将
RDD数据序列化后再保存在内存中，此时每个partition仅仅是一个字节数组而已，大大减少了对象数量，并
降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销，主要就是序列化与反序列化的开销。但是后
续算子可以基于纯内存进行操作，因此性能总体还是比较高的。此外，可能发生的问题同上，如果RDD中的数
据量过多的话，还是可能会导致OOM内存溢出的异常。
3、如果纯内存的级别都无法使用，那么建议使用MEMORY_AND_DISK_SER策略，而不是MEMORY_AND_DISK
策略。因为既然到了这一步，就说明RDD的数据量很大，内存无法完全放下。序列化后的数据比较少，可以节
省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中，内存缓存不下才会写入磁盘。
4、通常不建议使用DISK_ONLY和后缀为_2的级别：因为完全基于磁盘文件进行数据的读写，会导致性能急剧
降低，有时还不如重新计算一次所有RDD。后缀为_2的级别，必须将所有数据都复制一份副本，并发送到其他
节点上，数据复制以及网络传输会导致较大的性能开销，除非是要求作业的高可用性，否则不建议使用。

尽量避免使用Shuffle类算子

shuffle算子有可能导致数据倾斜！
join: reducejoin ＋ mapjoin

shuffle 过程中，各个节点上的相同 key 都会先写入本地磁盘文件中，然后其他节点需要通过网络传输
拉取各个节点上的磁盘文件中的相同 key。而且相同 key 都拉取到同一个节点进行聚合操作时，还有可
能会因为一个节点上处理的 key 过多，导致内存不够存放，进而溢写到磁盘文件中。因此在 shuffle 过
程中，可能会发生大量的磁盘文件读写的 IO 操作，以及数据的网络传输操作。磁盘 IO 和网络数据传输
也是 shuffle 性能较差的主要原因。

因此在我们的开发过程中，能避免则尽可能避免使用 reduceByKey、join、distinct、repartition 等会
进行 shuffle 的算子，尽量使用 map 类的非 shuffle 算子。这样的话，没有 shuffle 操作或者仅有较少
shuffle 操作的Spark 作业，可以大大减少性能开销。

Broadcast与map进行join代码示例

// 传统的join操作会导致shuffle操作。
// 因为两个RDD中，相同的key都需要通过网络拉取到一个节点上，由一个task进行join操作。
val rdd3 = rdd1.join(rdd2)
// Broadcast+map的join操作，不会导致shuffle操作。
// 使用Broadcast将一个数据量较小的RDD作为广播变量。
val rdd2Data = rdd2.collect()
val rdd2DataBroadcast = sc.broadcast(rdd2Data)
// 在rdd1.map算子中，可以从rdd2DataBroadcast中，获取rdd2的所有数据。
// 然后进行遍历，如果发现rdd2中某条数据的key与rdd1的当前数据的key是相同的，那么就判定可以进行
join。
// 此时就可以根据自己需要的方式，将rdd1当前数据与rdd2中可以连接的数据，拼接在一起（String或
Tuple）。
val rdd3 = rdd1.map(rdd2DataBroadcast...)
// 注意，以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用。
// 因为每个Executor的内存中，都会驻留一份rdd2的全量数据。

使用Map-Side预聚合的Shuffle操作

如果因为业务需要，一定要使用 shuffle 操作，无法用 map 类的算子来替代，那么尽量使用可以 mapside预聚合的算子。

所谓的 map-side 预聚合，说的是在每个节点本地对相同的 key 进行一次聚合操作，类似于
MapReduce 中的本地 combiner。map-side 预聚合之后，每个节点本地就只会有一条相同的 key，因
为多条相同的 key都被聚合起来了。其他节点在拉取所有节点上的相同 key 时，就会大大减少需要拉取
的数据数量，从而也就减少了磁盘 IO 以及网络传输开销。通常来说，在可能的情况下，建议使用
reduceByKey 或者 aggregateByKey 算子来替代掉 groupByKey 算子。因为 reduceByKey 和
aggregateByKey 算子都会使用用户自定义的函数对每个节点本地的相同key进行预聚合。而
groupByKey 算子是不会进行预聚合的，全量的数据会在集群的各个节点之间分发和传输，性能相对来
说比较差。

使用高性能的算子

除了 shuffle 相关的算子有优化原则之外，其他的算子也都有着相应的优化原则。

使用reduceByKey/aggregateByKey替代groupByKey

使用mapPartitions替代普通map

使用foreachPartitions替代foreach

使用filter之后进行coalesce操作

原理：如果原来有2n个分区，现在coalesce之后还有n个分区，底层逻辑就是：2个分区变成一个分区。
用 coalesce 减少 partition 数量，将 RDD 中的数据压缩到更少的 partition 之后，只要使用更少的 task 即可处理完所有的 partition。在某些场景下，对于性能的提升会有一定的帮助。

使用repartitionAndSortWithinPartitions替代repartition与sort类操作

广播大变量

在算子函数中使用到外部变量时，默认情况下，Spark 会将该变量复制多个副本，通过网络传输到 task
中，此时每个 task 都有一个变量副本。如果变量本身比较大的话（比如 100M，甚至 1G），那么大量
的变量副本在网络中传输的性能开销，以及在各个节点的 Executor 中占用过多内存导致的频繁 GC，都
会极大地影响性能。
因此对于上述情况，如果使用的外部变量比较大，建议使用 Spark 的广播功能，对该变量进行广播。广
播后的变量，会保证每个 Executor 的内存中，只驻留一份变量副本，而 Executor 中的 task 执行时共
享该 Executor 中的那份变量副本。这样的话，可以大大减少变量副本的数量，从而减少网络传输的性
能开销，并减少对 Executor 内存的占用开销，降低 GC 的频率。

广播大变量的代码示例

// 以下代码在算子函数中，使用了外部的变量。
// 此时没有做任何特殊操作，每个task都会有一份list1的副本。
val list1 = ...
rdd1.map(list1...)
// 以下代码将list1封装成了Broadcast类型的广播变量。
// 在算子函数中，使用广播变量时，首先会判断当前task所在Executor内存中，是否有变量副本。
// 如果有则直接使用；如果没有则从Driver或者其他Executor节点上远程拉取一份放到本地Executor内
存中。
// 每个Executor内存中，就只会驻留一份广播变量副本。
val list1 = ...
val list1Broadcast = sc.broadcast(list1)
rdd1.map(list1Broadcast...)

使用Kryo优化序列化性能

序列化和反序列化： java对象，如果要进行IO操作，就需要被序列化成字节数组
1、java中的序列化机制： 让自定义的类实现Serializable接口, 这种方式太重。！！
除了序列化属性的值意外，还会序列这个类的信息。
2、Hadoop当中：专门为hadoop提供了一种序列化机制：让自定义的类实现Writable
这两种方式，spark都不想用！
3、spark提供了一种新的优化方式：Kryo

在 Spark 中，主要有三个地方涉及到了序列化：

1、在算子函数中使用到外部变量时，该变量会被序列化后进行网络传输（见“原则七：广播大变量”中的讲
解）。
2、将自定义的类型作为 RDD 的泛型类型时（比如JavaRDD，Student是自定义类型），所有自定义类型对
象，都会进行序列化。因此这种情况下，也要求自定义的类必须实现Serializable接口。
3、使用可序列化的持久化策略时（比如MEMORY_ONLY_SER），Spark会将RDD中的每个partition都序列
化成一个大的字节数组。

对于这三种出现序列化的地方，我们都可以通过使用 Kryo 序列化类库，来优化序列化和反序列化的性
能。Spark 默认使用的是 Java 的序列化机制，也就是 ObjectOutputStream/ObjectInputStream API
来进行序列化和反序列化。但是 Spark 同时支持使用 Kryo序列化库，Kryo 序列化类库的性能比 Java
序列化类库的性能要高很多。官方介绍，Kryo 序列化机制比 Java 序列化机制，性能高 10 倍左右。
Spark 之所以默认没有使用 Kryo 作为序列化类库，是因为 Kryo 要求最好要注册所有需要进行序列化的
自定义类型，因此对于开发者来说，这种方式比较麻烦。

// 创建SparkConf对象。
val conf = new SparkConf().setMaster(...).setAppName(...)
// 设置序列化器为KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注册要序列化的自定义类型。
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
**// 提供一种最暴力的方案：只要是RDD的泛型类，都应该使用这种方式**

优化数据结构

Java 中，有三种类型比较耗费内存：
1、对象，每个Java对象都有对象头、引用等额外的信息，因此比较占用内存空间。
2、字符串，每个字符串内部都有一个字符数组以及长度等额外信息。
3、集合类型，比如HashMap、LinkedList等，因为集合类型内部通常会使用一些内部类来封装集合元素，
比如Map.Entry。

因此 Spark 官方建议，在 Spark 编码实现中，特别是对于算子函数中的代码，尽量不要使用上述三种
数据结构，尽量使用字符串替代对象，使用原始类型（比如 Int、Long）替代字符串，使用数组替代集
合类型，这样尽可能地减少内存占用，从而降低 GC 频率，提升性能。