spark 基础知识整理（二）- RDD专题

2019年6月8日 210次阅读 来源: Graceleeman

一、RDD及其特点

**1. **RDD（Resillient Distributed Dataset）弹性分布式数据集，是spark提供的核心抽象。它代表一个不可变、可分区、里面的元素可并行计算的集合
**2. **RDD在抽象上来说是一种元素集合，包含了数据。它是被分区的，分为多个分区，每个分区分布在集群中的不同节点上，从而让RDD中的数据可以被并行操作（分布式数据集）
**3. **RDD通常通过hadoop上的文件，即hdfs文件或者hive表来进行创建，有时也可以通过应用程序中的集合来创建。
**4. **RDD最重要的特性就是提供了容错性，可以自动从节点失败中恢复过来，即某节点上的RDD partition，因为节点故障，导致数据丢了，那么RDD会自动通过自己数据来源重新计算该partition，这一切对使用者是透明的
**5. **RDD的数据默认情况下存在内存中，但是在内存资源不足是，spark会自动将RDD数据写入磁盘（弹性）
注意：RDD的每个partition，在spark节点上存储时，默认都是放在内存中的，但是如果说内存放不下这么多数据，比如每个节点最多放5w数据，结果每个partition市10w数据，那么就会把partition中的部分数据写入磁盘，进行保存。而上述这一切，对于用户来说，都是完全透明的，也就是不用去管RDD的数据存放在内存还是磁盘，只要关注你针对RDD来进行计算和处理等操作即可。所以说RDD这种自动进行内存和磁盘之间权衡和切换的机制，就是RDD的弹性的特点所在。

二、RDD的属性

1. partitions(分区)。每个RDD包括多个分区, 这既是RDD的数据单位, 也是计算粒度, 每个分区是由一个Task线程处理. 在RDD创建的时候可以指定分区的个数, 如果没有指定, 那么默认分区的个数是CPU的核数（standalone）.
每一分区对应一个内存block, 由BlockManager分配.
2. partitioner(分区方法)。这个属性指的是RDD的partitioner函数(分片函数), 分区函数就是将数据分配到指定的分区, 这个目前实现了HashPartitioner和RangePartitioner, 只有key-value的RDD才会有分片函数, 否则为none.。分片函数不仅决定了当前分片的个数, 同时决定parent shuffle RDD的输出的分区个数。
3. dependencies(依赖关系)。Spark的运行过程就是RDD之间的转换, 因此, 必须记录RDD之间的生成关系(新RDD是由哪个或哪几个父RDD生成), 这就是所谓的依赖关系, 这样既有助于阶段和任务的划分, 也有助于在某个分区出错的时候, 只需要重新计算与当前出错的分区有关的分区,而不需要计算所有的分区。
窄依赖：父 RDD 的 partition 至多被一个子 RDD partition 依赖（OneToOneDependency，RangeDependency）
宽依赖：父 RDD 的 partition 被多个子 RDD partitions 依赖（ShuffleDependency）

《spark 基础知识整理（二）- RDD专题》

4. compute(获取分区迭代列表)。计算属性: 当调用 RDD#iterator 方法无法从缓存或checkpoint中获取指定 partition 的迭代器时，就需要调用 compute 方法来获取RDD不仅包含有数据, 还有在数据上的计算, 每个RDD以分区为计算粒度, 每个RDD会实现compute函数, compute函数会和迭代器(RDD之间转换的迭代器)进行复合, 这样就不需要保存每次compute运行的结果.
5. preferedLocations(优先分配节点列表)。对于分区而言返回数据本地化计算的节点列表。也就是说, 每个RDD会报出一个列表（Seq）, 而这个列表保存着分片优先分配给哪个Worker节点计算, spark坚持移动计算而非移动数据的原则. 也就是尽量在存储数据的节点上进行计算。要注意的是，并不是每个 RDD 都有 preferedLocation，比如从 Scala 集合中创建的 RDD 就没有，而从 HDFS 读取的 RDD 就有。

三、创建RDD

进行spark核心编程时，先要创建一个初始的RDD，该RDD中，通常就代表和包含了spark应用程序的输入源数据，然后在创建了初识的RDD之后，才可以通过spark core提供的transformation算子，对RDD进行转换，来获取其他的RDD。

spark core提供了二种创建RDD的方式：

  1. 由已经存在的Scala集合创建RDD
先要启动spark-shell
scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8,9,10),2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:24
求和
scala> val sum = rdd1.reduce(_+_)
sum: Int = 55
  1. 由外部存储系统的数据集创建
使用本地文件创建RDD  
使用HDFS文件创建RDD 
scala> val rdd3 = sc.textFile("hdfs://master:9000/student/2016113012/spark/words.txt")
rdd3: org.apache.spark.rdd.RDD[String] = hdfs://master:9000/student/2016113012/spark/words.txt MapPartitionsRDD[46] at textFile at <console>:24
统计文本字数
scala> val count = rdd3.map(lines => lines.length()).reduce(_+_)
count: Int = 48
文件内容如下
hello scala
hello java
hello python
hello wujiadong

spark默认会为hdfs的每一个block创建一个partition，但是也可以通过textFile()的第二个参数手动设置分区数量，只能比block数量多，不能比他更少。

四、RDD运行流程

RDD在Spark中运行大概分为以下三步：

  1. 创建RDD对象
  2. DAGScheduler模块介入运算，计算RDD之间的依赖关系，RDD之间的依赖关系就形成了DAG

  3. 每一个Job被分为多个Stage。划分Stage的一个主要依据是当前计算因子的输入是否是确定的，如果是则将其分在同一个Stage，避免多个Stage之间的消息传递开销

    《spark 基础知识整理（二）- RDD专题》

    以下面一个按 A-Z 首字母分类，查找相同首字母下不同姓名总个数的例子来看一下 RDD 是如何运行起来的 《spark 基础知识整理（二）- RDD专题》

五、RDD如何操作

对于RDD可以有两种操作算子：
1. 转换（Transformation）：Transformation操作是延迟计算的，也就是说从一个RDD转换生成另一个RDD的转换操作不是马上执行，需要等到有Action操作的时候才会真正触发运算。
2. 行动（Action）：Action算子会触发Spark提交作业（Job），并将数据输出Spark系统。

1.Transformation具体内容：

《spark 基础知识整理（二）- RDD专题》

2.Action具体内容：

《spark 基础知识整理（二）- RDD专题》

六、RDD的优缺点

优点:

  1. 编译时类型安全, 编译时就能检查出类型错误。
  2. 面向对象的编程风格, 直接通过类名点的方式来操作数据。

缺点:

  1. 序列化和反序列化的性能开销, 无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化。
  2. GC的性能开销, 频繁的创建和销毁对象, 势必会增加GC。
    原文作者：Graceleeman
    原文地址: https://www.jianshu.com/p/7a8d5ee1bc44
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞