Spark性能调优：RDD的复用及持久化

避免创建重复的 RDD

通常来说，开发一个 Spark 作业时，首先是基于某个数据源（比如 Hive 表或 HDFS 文件）创建一个初始的 RDD；接着对这个 RDD 执行某个算子操作，然后得到下一个 RDD；以此类推，循环往复，直到计算出最终我们需要的结果。在这个过程中，多个 RDD 会通过不同的算子操作（比如 map、reduce 等）串起来，这个 “RDD 串”，就是 RDD lineage，也就是 “RDD 的血缘关系链”。

我们在开发过程中要注意：对于同一份数据，只应该创建一个 RDD，不能创建多个 RDD 来代表同一份数据。

在开发 RDD lineage 极其冗长的 Spark 作业时，可能会忘了自己之前对于某一份数据已经创建过一个 RDD 了，从而导致对于同一份数据，创建了多个 RDD。这就意味着，我们的 Spark 作业会进行多次重复计算来创建多个代表相同数据的 RDD，进而增加了作业的性能开销。

一个简单的例子：

<span style="color:#000000"><code>// 需要对名为“hello.txt”的HDFS文件进行一次map操作，再进行一次reduce操作。也就是说，需要对一份数据执行两次算子操作。

// 错误的做法：对于同一份数据执行多次算子操作时，创建多个RDD。
// 这里执行了两次textFile方法，针对同一个HDFS文件，创建了两个RDD出来，然后分别对每个RDD都执行了一个算子操作。
// 这种情况下，Spark需要从HDFS上两次加载hello.txt文件的内容，并创建两个单独的RDD；第二次加载HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的。
val rdd1 = sc.textFile(<span style="color:#009900 !important">"hdfs://192.168.0.0:8020/hello.txt"</span>)
rdd1.map(<span style="color:#000088 !important">...</span>)
val rdd2 = sc.textFile(<span style="color:#009900 !important">"hdfs://192.168.0.0:8020/hello.txt"</span>)
rdd2.reduce(<span style="color:#000088 !important">...</span>)

// 正确的用法：对于一份数据执行多次算子操作时，只使用一个RDD。
// 这种写法很明显比上一种写法要好多了，因为我们对于同一份数据只创建了一个RDD，然后对这一个RDD执行了多次算子操作。
// 但是要注意到这里为止优化还没有结束，由于rdd1被执行了两次算子操作，第二次执行reduce操作的时候，还会再次从源头处重新计算一次rdd1的数据，因此还是会有重复计算的性能开销。
// 要彻底解决这个问题，必须结合“原则三：对多次使用的RDD进行持久化”，才能保证一个RDD被多次使用时只被计算一次。
val rdd1 = sc.textFile(<span style="color:#009900 !important">"hdfs://192.168.0.0:8020/hello.txt"</span>)
rdd1.map(<span style="color:#000088 !important">...</span>)
rdd1.reduce(<span style="color:#000088 !important">...</span>)</code></span>
           

尽可能复用同一个 RDD

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

一个简单的例子

<span style="color:#000000"><code>// 错误的做法。

// 有一个<Long, String>格式的RDD，即rdd1。
// 接着由于业务需要，对rdd1执行了一个map操作，创建了一个rdd2，而rdd2中的数据仅仅是rdd1中的value值而已，也就是说，rdd2是rdd1的子集。
JavaPairRDD<Long, String> rdd1 = <span style="color:#000088 !important">...</span>
JavaRDD<String> rdd2 = rdd1.map(<span style="color:#000088 !important">...</span>)

// 分别对rdd1和rdd2执行了不同的算子操作。
rdd1.reduceByKey(<span style="color:#000088 !important">...</span>)
rdd2.map(<span style="color:#000088 !important">...</span>)

//-------------------------------------------
// 正确的做法。

// 上面这个case中，其实rdd1和rdd2的区别无非就是数据格式不同而已，rdd2的数据完全就是rdd1的子集而已，却创建了两个rdd，并对两个rdd都执行了一次算子操作。
// 此时会因为对rdd1执行map算子来创建rdd2，而多执行一次算子操作，进而增加性能开销。

// 其实在这种情况下完全可以复用同一个RDD。
// 我们可以使用rdd1，既做reduceByKey操作，也做map操作。
// 在进行第二个map操作时，只使用每个数据的tuple._2，也就是rdd1中的value值，即可。
JavaPairRDD<Long, String> rdd1 = <span style="color:#000088 !important">...</span>
rdd1.reduceByKey(<span style="color:#000088 !important">...</span>)
rdd1.map(tuple._2...)

// 第二种方式相较于第一种方式而言，很明显减少了一次rdd2的计算开销。
// 但是到这里为止，优化还没有结束，对rdd1我们还是执行了两次算子操作，rdd1实际上还是会被计算两次。
// 因此还需要配合“对多次使用的RDD进行持久化”进行使用，才能保证一个RDD被多次使用时只被计算一次。
</code></span>
           

对多次使用的 RDD 进行持久化

Spark 中对于一个 RDD 执行多次算子的默认原理是这样的：每次你对一个 RDD 执行一个算子操作时，都会重新从源头处计算一遍，计算出那个 RDD 来，然后再对这个 RDD 执行你的算子操作。这种方式的性能是很差的。

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

<span style="color:#000000"><code>// 如果要对一个RDD进行持久化，只要对这个RDD调用cache()和persist()即可。

// 正确的做法。
// cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。
// 此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。
// 第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。
val rdd1 = sc.textFile(<span style="color:#009900 !important">"hdfs://192.168.0.1:9000/hello.txt"</span>).cache()
rdd1.map(<span style="color:#000088 !important">...</span>)
rdd1.reduce(<span style="color:#000088 !important">...</span>)

// persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化。
// 比如说，StorageLevel.MEMORY_AND_DISK_SER表示，内存充足时优先持久化到内存中，内存不充足时持久化到磁盘文件中。
// 而且其中的_SER后缀表示，使用序列化的方式来保存RDD数据，此时RDD中的每个partition都会序列化成一个大的字节数组，然后再持久化到内存或磁盘中。
// 序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，从而发生频繁GC。
val rdd1 = sc.textFile(<span style="color:#009900 !important">"hdfs://192.168.0.1:9000/hello.txt"</span>).persist(StorageLevel.MEMORY_AND_DISK_SER)
rdd1.map(<span style="color:#000088 !important">...</span>)
rdd1.reduce(<span style="color:#000088 !important">...</span>)</code></span>
           

对于 persist () 方法而言，我们可以根据不同的业务场景选择不同的持久化级别。

持久化级别	含义
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 计算时还可以使用该数据在其他节点上的副本。如果没有副本的话，就只能将这些数据从源头处重新计算一遍了。

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

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