第四章 Spark编程进阶思维导图

弓长-张

2024-05-08

浏览量: 6

设置数据分区

设置RDD持久化

存储数据

Spark编程进阶功能介绍

树图思维导图提供《第四章 Spark编程进阶》在线思维导图免费制作，点击“编辑”按钮，可对《第四章 Spark编程进阶》进行在线思维导图编辑，本思维导图属于思维导图模板主题，文件编号是：86ba5a1a95af6c3ad767f9365ec61b1b

思维导图大纲

设置RDD持久化

Spark RDD 是惰性求值的,因此如果需要多次使用同一个 RDD,那么调用时每次都需要计算该 RDD 并执行它依赖的其他转换操作。在需要多次迭代的计算中，由于常常需要多次使用同一组数据，因此计算资源消耗会非常大。为了避免多次计算同一个 RDD，可以在 Spark 中设置数据持久化。

每个 RDD 一般情况下是由多个分区组成的，RDD的数据分布在多个节点中，因此 Spark持久化RDD数据时，由参与计算该RDD的节点各自保存所求出的分区数据。持久化 RDD后，再使用该 RDD时将不需要重新计算，直接获取各分区保存好的数据即可。如果其中有一个节点因为某种原因出现故障，那么 Spark 不需要计算所有分区的数据，只需要重新计算丢失的分区数据即可。如果希望节点故障不影响执行速度，那么在内存充足的情况下，可以使用双副本保存的高可靠机制，当其中一个副本有分区数据丢失时，则会从另一个副本中读取数据。

RDD 的持久化操作有 cache()和 persist()两种方法。每一个 RDD 都可以用不同的存储级别进行保存，从而允许持久化数据集在硬盘或内存中作为序列化的Java对象,甚至跨节点复制。存储级别是通过 org.apache.spark.storage.StorageLevel对象确定的,常用存储级别如表 4-5所示。cache()方法是使用默认存储级别的快捷方法，也就是StorageLevel.MEMORY ONLY将反序列化的对象存人内存)。如果需要将数据保存到1个副本中，那么可以通过在存储级引末尾加上“_2”将数据持久化并保存为2份。

RDD的存储级别应该根据需要以及具体情况设置，在 RDD 参与第一次计算后，RDD会根据设置的存储级别保存计算后的值到内存或磁盘中。只有未曾设置存储级别的 RDD才能设置存储级别，已经设置了存储级别的 RDD不能修改存储级别。针对将数据存储到内存中的存储策略，如果内存不足,那么 Spark将使用最近最少使用(Least Recently Used，LRU)缓存策略清除最老的分区，为新的 RDD 提供存储空间。因此，缓存在内存中的 RDD 分区数据是会被清除的，不能长久保存，而缓存在磁盘上的数据则不会被清除。若内存不够存放 RDD 所有分区数据，则该 RDD 数据可能不会进行持久化，下次对该 RDD 执行转换操作或行动操作时，没有被持久化的 RDD 数据需要重新计算。如果需要人为清除已经不需要的数据，可以使用 RDD 提供的 unpersist()方法。

设置数据分区

Spark RDD是多个分区组成的数据集合，在分布式程序中，通信的代价是很大的，因此控制数据分区、减少网络传输是提高程序运行整体性能的重要方式。只有键值对 RDD 才能设置分区方式，非键值对RDD分区的值是 None。系统是根据一个针对键的函数对元素进行分区的，虽然不能控制每个键具体划分到哪个节点,但是可以控制相同特性的键落在同-个分区。每个 RDD 的分区 ID 范围是 0~(numPartitions-1)，numPartitions 是分区的个数。

设置分区方式使用的是 partitionByO方法，需要传人一种分区方式作为参数。例如。对键值对 RDD demordd 设置分区方式，可以通过“demordd.partitionBy(分区方式)”的方式完成。获取 RDD 的分区可以使用 RDD的 partitioner()方法，该方法会返回一个 scala.Option对象，可以通过这个 Option 对象调用 isDefined 查看该分区中是否有值，并用 get()方法获取分区中的值，值是一个 spark.Partitioner 对象。

Spark 的系统分区方式有两种，一种是哈希分区(HashPartitioner)，根据哈希值分区;另一种是范围分区(RangePartitioner )，将一定范围的数据映射到一个分区中。用户也可以通过自定义分区器完成特定的分区要求。

实现自定义分区器，需要继承 org.apache.spark,Partitioner 类并实现其中的3个方法。 (1)defnumPartitions:Int:返回需要创建的分区个数。 (2)def getPartition(key: Any):需要对输人的键进行处理，并返回该键的分区 ID，范围一定是0~(numPartitions-1)。 (3)eguals(other: Any):Java 语言中判断两个数据是否相等的方法，之所以要求用户实现 equals（）方法，是因为 Spark 内部会比较两个 RDD 的分区是否一样。

getPartition0方法中，对 RDD 的键进行判断，若为偶数则存储在0分区中，否则存储在1分区中。equals0方法比较通过自定义的 MyPartition 分区器得到的分区是否与其他RDD 分区一样，以此判断两个 RDD 的分区方式是否相同。

自定义分区器可以控制数据的分区个数、每个分区中数据的内容。在存储数据时，数据按照分区器的设置保存在不同的文件中。由于RDD的分区存储特性，如果不自定义分区方法，或数据不是键值对 RDD，那么数据可能会分布在几个分区中，最终结果也会被存储在几个文件中，不利于对所有数据结果的查看。

RDD 中还有两种可以不自定义分区器但也可以对任何类型 RDD 进行简单重分区的方法，即 coalesce0方法和 repartition()方法。 (1)coalesce(numPartitions: Int,shuffle: Boolean=false)方法使用哈希分区方式对 RDD进行重分区。该方法支持输入两个参数，第一个参数为重分区的数目，第二个参数为是否进行 shuffle，默认为 false。shuffle 为 false 时，重设分区个数只能比 RDD 原有分区数小，如果要重设的分区个数大于 RDD 原有的分区数，那么 RDD的分区数将不变;如果shufle 为 true，那么重设的分区个数不管比原有的 RDD分区数大或小，RDD 都可以重设分区个数。 (2)repartition(numPartitions:Int)方法本质上是 coalesce()方法的第二个参数 shufle 为true 的简单实现。