spark 基础知识整理(二)- RDD专题
一、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)
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的方式:
- 由已经存在的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
- 由外部存储系统的数据集创建
使用本地文件创建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中运行大概分为以下三步:
- 创建RDD对象
- DAGScheduler模块介入运算,计算RDD之间的依赖关系,RDD之间的依赖关系就形成了DAG
-
每一个Job被分为多个Stage。划分Stage的一个主要依据是当前计算因子的输入是否是确定的,如果是则将其分在同一个Stage,避免多个Stage之间的消息传递开销
以下面一个按 A-Z 首字母分类,查找相同首字母下不同姓名总个数的例子来看一下 RDD 是如何运行起来的
五、RDD如何操作
对于RDD可以有两种操作算子:
1. 转换(Transformation):Transformation操作是延迟计算的,也就是说从一个RDD转换生成另一个RDD的转换操作不是马上执行,需要等到有Action操作的时候才会真正触发运算。
2. 行动(Action):Action算子会触发Spark提交作业(Job),并将数据输出Spark系统。
1.Transformation具体内容:
2.Action具体内容:
六、RDD的优缺点
优点:
- 编译时类型安全, 编译时就能检查出类型错误。
- 面向对象的编程风格, 直接通过类名点的方式来操作数据。
缺点:
- 序列化和反序列化的性能开销, 无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化。
- GC的性能开销, 频繁的创建和销毁对象, 势必会增加GC。