Spark大数据分析与实战笔记（第三章 Spark RDD 弹性分布式数据集-04）

每日一句正能量

勤奋踏实，诚实守信。我始终相信我读过的所有书都不会白读，它总会在未来日子的某一个场合帮助我表现得更出色，读书是可以给人以力量的，它更能给人快乐。

第3章 Spark RDD弹性分布式数据集

章节概要

传统的MapReduce虽然具有自动容错、平衡负载和可拓展性的优点，但是其最大缺点是采用非循环式的数据流模型，使得在迭代计算式要进行大量的磁盘IO操作。Spark中的RDD可以很好的解决这一缺点。

RDD是Spark提供的最重要的抽象概念，我们可以将RDD理解为一个分布式存储在集群中的大型数据集合，不同RDD之间可以通过转换操作形成依赖关系实现管道化，从而避免了中间结果的I/O操作，提高数据处理的速度和性能。接下来，本章将针对RDD进行详细讲解。

3.6 RDD机制

Spark为RDD提供了两个重要的机制，分别是持久化机制(即缓存机制)和容错机制。接下来，本节将针对持久化机制和容错机制进行详细介绍。

3.6.1 持久化机制

• 在Spark中，RDD是采用惰性求值，即每次调用行动算子操作，都会从头开始计算。然而，每次调用行动算子操作，都会触发一次从头开始的计算，这对于迭代计算来说，代价是很大的，因为迭代计算经常需要多次重复的使用同一组数据集，所以，为了避免重复计算的开销，可以让Spark对数据集进行持久化。
• 通常情况下，一个RDD是由多个分区组成的，RDD中的数据分布在多个节点中，因此，当持久化某个RDD时，每一个节点都将把计算分区的结果保存在内存中，若对该RDD或衍生出的RDD进行其他行动算子操作时，则不需要重新计算，直接去取各个分区保存数据即可，这使得后续的行动算子操作速度更快（通常超过10倍)，并且缓存是Spark构建迭代式算法和快速交互式查询的关键。
• RDD的持久化操作有两种方法，分别是cache()方法和persist()方法。每一个持久化的RDD都可以使用不同的存储级别存储，从而允许持久化数据集在硬盘或者内存作为序列化的Java对象，甚至可以跨节点复制。

1. persist()方法的存储级别是通过StorageLevel对象(Scala、Java、Python）设置的。
2. cache()方法的存储级别是使用默认的存储级别(即StorageLevel.MEMORY_ONLY(将反序列化的对象存入内存)）。
   接下来，通过—张表介绍一下持久化RDD的存储级别。

为了大家更好地理解，接下来，通过代码演示如何使用persist()方法和cache()方法对RDD进行持久化。

1. 使用persist()方法对RDD进行持久化
   定义一个列表list，通过该列表创建一个RDD，然后通过persist持久化操作和算子操作统计RDD中的元素个数以及打印输出RDD中的所有元素。具体代码如下：

scala> import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel

scala> val list=List("hadoop","spark","hive")
list: List[String] = List(hadoop, spark, hive)

scala> val listRDD=sc.parallelize(list)
listRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:29

scala> listRDD.persist(StorageLevel.DISK_ONLY)
res0: listRDD.type = ParallelCollectionRDD[0] at parallelize at <console>:29

scala> println(listRDD.count())
3

scala> println(listRDD.collect().mkString(","))
hadoop,spark,hive
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

listRDD.persist(StorageLevel.DISK_ONLY)设置了持久化级别，但是这行代码并不会马上执行持久化操作，只有当第一次调用行动算子后（println(listRDD.count())），才会去挂靠持久化，而第2次调用算子时，使用已经产生的持久化RDD，速度会比之前快。

2. 使用cache()方法对RDD进行持久化
   定义一个列表list，通过该列表创建一个RDD，然后通过cache持久化操作和算子操作统计RDD中的元素个数以及打印输出rdd中的所有元素。具体代码如下：

scala> val list=List("hadoop","spark","hive")
list: List[String] = List(hadoop, spark, hive)

scala> val listRDD=sc.parallelize(list)
listRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[1] at parallelize at <console>:29

scala> listRDD.cache()
res3: listRDD.type = ParallelCollectionRDD[1] at parallelize at <console>:29

scala> println(listRDD.count())
3

scala> println(listRDD.collect().mkString(","))
hadoop,spark,hive
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3.6.2 容错机制

当Spark集群中的某一个节点由于宕机导致数据丢失，则可以通过Spark中的RDD进行容错恢复已经丢失的数据。RDD提供了两种故障恢复的方式，分别是血统（Lineage）方式和设置检查点（checkpoint）方式。

• 血统(Lineage)方式，主要是根据RDD之间的依赖关系对丢失数据的RDD进行数据恢复。如果丢失数据的子RDD在进行窄依赖运算r则只需要把丢失数据的父RDD的对应分区进行重新计算即可，不需要依赖其他的节点，并且在计算过程中不会存在冗余计算;若丢失数据的RDD进行宽依赖运算，则需要父RDD的所有分区都要进行从头到尾的计算，在计算过程中会存在冗余计算。为了解决宽依赖运算中出现的计算冗余问题，Spark又提供了另一种方式进行数据容错，即设置检查点(checkpoint)方式。

• 设置检查点(checkPoint)方式，本质上是将RDD写入磁盘进行存储。当RDD在进行宽依赖运算时，只需要在中间阶段设置一个检查点进行容错，即通过Spark中的sparkContext对象调用setCheckpoint()方法，设置一个容错文件系统目录(如HDFS）作为检查点checkpoint，将checkpoint的数据写入之前设置的容错文件系统中进行高可用的持久化存储，若是后面有节点出现宕机导致分区数据丢失，则可以从做检查点的RDD开始重新计算即可，不需要进行从头到尾的计算，这样就会减少开销。

总结

RDD（Resilient Distributed Dataset，弹性分布式数据集）是Spark中的核心概念之一。RDD具有分布式和不可变的特性，能够在集群中进行并行计算。RDD机制通过持久化机制和容错机制来保证数据的可靠性和高效性。

持久化机制是指RDD可以将计算结果缓存在内存中，以便在后续的计算中复用，减少了数据的重复计算和IO开销。RDD提供了多种持久化级别，包括内存和磁盘，用户可以根据实际需求选择不同的级别。

容错机制是指RDD具有自动容错和恢复能力。RDD通过将数据划分成一系列的分区（partitions）来实现容错。每个分区都会在集群中的不同节点上进行备份，一旦某个节点发生故障，Spark可以自动将该分区计算任务转移到其他可用节点上进行处理，从而确保整个计算过程不会中断。

总之，RDD机制通过持久化机制和容错机制来提高了计算效率和可靠性，并且可以在分布式环境中灵活应用。这让Spark成为了一个高性能和可靠的大数据处理框架。

转载自：https://blog.csdn.net/u014727709/article/details/132509716
欢迎