大数据笔记(二十九)——RDD简介特性及常用算子

Posted lingluo2017

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了大数据笔记(二十九)——RDD简介特性及常用算子相关的知识,希望对你有一定的参考价值。

1、什么是RDD? 最核心
(*)弹性分布式数据集,Resilent distributed DataSet
(*)Spark中数据的基本抽象
(*)结合源码,查看RDD的概念

RDD属性
* Internally, each RDD is characterized by five main properties:
*
* - A list of partitions
一组分区,把数据分成了的不同的分区,每个分区可能运行在不同的worker

 


* - A function for computing each split
一个函数,用于计算每个分区中的数据
RDD的函数(算子)
(1)Transformation
(2)Action

* - A list of dependencies on other RDDs
RDD之间依赖关系:(1)窄依赖 (2)宽依赖
根据依赖的关系,来划分任务的Stage(阶段)

* - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
* - Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

如何创建一个RDD?有两种方式
(1)使用sc.parallelize方法

val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8),3)

(2)通过使用外部的数据源创建RDD:比如:HDFS

val rdd2 = sc.textFile("hdfs://bigdata11:9000/input/data.txt")
val rdd2 = sc.textFile("/root/temp/input/data.txt")

2、Transformation算子:不会触发计算、延时加载(lazy值)

RDD API网址:http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.rdd.RDD

常见的RDD算子:

map(func):该操作是对原来的RDD进行操作后,返回一个新的RDD
filter: 过滤操作、返回一个新的RDD
flatMap:类似map
mapPartitions:对每个分区进行操作
mapPartitionsWithIndex: 对每个分区进行操作,带分区的下标
union 并集
intersection 交集
distinct 去重
groupByKey: 都是按照Key进行分组
reduceByKey: 都是按照Key进行分组、会有一个本地操作(相当于:Combiner操作)


3、Action算子:会触发计算

collect: 触发计算、打印屏幕上。以数组形式返回
count: 求个数
first: 第一个元素(take(1))
take(n)
saveAsTextFile: 会转成String的形式,会调用toString()方法
foreach: 在RDD的每个元素上进行某个操作

 


4、RDD的缓存机制:默认在内存中
(*)提高效率
(*)默认:缓存在Memory中
(*)调用:方法:persist或者cache

def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)
def cache(): this.type = persist()

(*)缓存的位置:StorageLevel定义的

val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

 (*)示例: 

测试数据:Oracle数据库的订单变 sales表(大概92万)
步骤
(1)从HDFS读入数据

val rdd1 = sc.textFile("hdfs://bigdata11:9000/input/sales")

(2)计算

rdd1.count ---> Action,这一次没有缓存
rdd1.cache ---> 缓存数据,但是不会触发计算,cache是一个Transformation
rdd1.count ----> 触发计算,将结果缓存
rdd1.count ----> ???会从哪里得到数据

通过UI进行监控:

 

IDEA功能键:ctrl + n 查找类
ctl+alt+shit+N 在类中找方法

5、RDD的容错机制:checkpoint检查点
(1)复习检查点:HDFS中,合并元信息
                              Oracle中,会以最高优先级唤醒数据库写进程(DBWn),来内存中的脏数据---> 数据文件
(2)RDD的检查点:容错机制,辅助Lineage(血统)---> 整个计算的过程
如果lineage越长,出错的概率就越大。出错之后,从最近一次的检查点开始运行。

两种类型

(1)本地目录 : 需要将spark-shell运行在本地模式上      

 


(2)HDFS目录: 需要将spark-shell运行在集群模式上

scala> sc.setCheckpointDir("hdfs://bigdata11:9000/spark/checkpoint")

scala> val rdd1 = sc.textFile("hdfs://bigdata11:9000/input/sales")
rdd1: org.apache.spark.rdd.RDD[String] = hdfs://bigdata11:9000/input/sales MapPartitionsRDD[41] at textFile at <console>:24

scala> rdd1.checkpoint

scala> rdd1.count

源码中对于检查点的说明:

/**
* Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint
* directory set with `SparkContext#setCheckpointDir` and all references to its parent
* RDDs will be removed. This function must be called before any job has been
* executed on this RDD. It is strongly recommended that this RDD is persisted in
* memory, otherwise saving it on a file will require recomputation.
*/    

RDD的检查点多久发出一次,是手动发出的吗?
1、不是手动
2、每个RDD计算完成后

查看源码

/**
* Performs the checkpointing of this RDD by saving this. It is called after a job using this RDD
* has completed (therefore the RDD has been materialized and potentially stored in memory).
* doCheckpoint() is called recursively on the parent RDDs.
*/
private[spark] def doCheckpoint(): Unit = {

 


6、RDD的依赖关系、划分Spark任务的Stage(阶段)
(*)窄依赖(Narrow Dependencies):每一个父RDD的分区最多被子RDD的一个分区使用
       比方:独生子女

  举例:map,filter,union

(*)宽依赖(Wide Dependencies):多个子RDD的分区会依赖同一个父RDD的分区
         比方:超生

  举例:groupByKey

  比如分区1和分区2都有10号部门的员工,那么在统计10号部门(key)的员工时需要依赖分区1和分区2,它们属于不同的父RDD的分区。

根据宽依赖和窄依赖的标准,我们可以划分任务的Stage(阶段)


7、RDD算子的基础例子

1、创建一个RDD(数字)
    val rdd1 = sc.parallelize(List(5,6,1,2,10,4,12,20,100,30))
    
    每个元素*2,然后排序
    val rdd2 = rdd1.map(_*2).sortBy(x=>x,true)
完整 val rdd2
= rdd1.map((x:Int)=>x*2) 过滤出大于10的元素 val rdd3 = rdd2.filter(_>10) rdd3.collect 2、创建一个RDD(字符) val rdd1 = sc.parallelize(Array("a b c","d e f","h i j")) val rdd2 = rdd1.flatMap(_.split(\' \')) rdd2.collect 3、集合运算、去重 val rdd1 = sc.parallelize(List(5,6,7,8,1,2)) val rdd2 = sc.parallelize(List(1,2,3,4)) val rdd3 = rdd1.union(rdd2) rdd3.distinct.collect val rdd4 = rdd1.intersection(rdd2)
4、分组
    val rdd1 = sc.parallelize(List(("Tom",1000),("Jerry",3000),("Mary",2000)))
    val rdd2 = sc.parallelize(List(("Jerry",500),("Tom",3000),("Mike",2000)))
    
    并集
    val rdd3 = rdd1 union rdd2
    scala> val rdd4 = rdd3.groupByKey
    rdd4: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[27] at groupByKey at <console>:30

    scala> rdd4.collect
    res8: Array[(String, Iterable[Int])] = Array((Tom,CompactBuffer(1000, 3000)),
                                                 (Jerry,CompactBuffer(3000, 500)), 
                                                 (Mike,CompactBuffer(2000)), 
                                                 (Mary,CompactBuffer(2000)))

 

 

六、Spark RDD的高级算子
1、mapPartitionsWithIndex: 对RDD中的每个分区进行操作,带有分区号
定义:def mapPartitionsWithIndex[U](f: (Int, Iterator[T])=>Iterator[U], preservesPartitioning: Boolean = false)
(implicit arg0: ClassTag[U]): RDD[U]
参数说明:
f: (Int, Iterator[T])=>Iterator[U]
(*)Int: 分区号
(*)Iterator[T]: 该分区中的每个元素
(*)返回值:Iterator[U]

Demo:
(1)创建一个RDD:

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),2)

(2)创建一个函数,作为f的值

def func1(index:Int,iter:Iterator[Int]):Iterator[String] ={
iter.toList.map(x=>"[PartID:" + index +",value="+x+"]").iterator
}

(3)调用

rdd1.mapPartitionsWithIndex(func1).collect

(4)结果:

Array([PartID:0,value=1], [PartID:0,value=2], [PartID:0,value=3], [PartID:0,value=4], 
[PartID:1,value=5], [PartID:1,value=6], [PartID:1,value=7], [PartID:1,value=8], [PartID:1,value=9])    

 


2、aggregate:聚合操作
定义:def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
作用:先对局部进行操作,再对全局进行操作

 


举例:

val rdd1 = sc.parallelize(List(1,2,3,4,5),2)

(1)求每个分区最大值的和
先查看每个分区中的元素:

rdd1.mapPartitionsWithIndex(func1).collect

rdd1.aggregate(0)(math.max(_,_),_+_)

(2)改一下:

rdd1.aggregate(0)(_+_,_+_) ====> 15 两个分区求和并相加
rdd1.aggregate(10)(math.max(_,_),_+_) ===> 30 初始值是10,每个分区里有10,初始值10+分区一10+分区二10 = 30

(3)一个字符串的例子

 

3、aggregateByKey

(1)类似aggregate,也是先对局部,再对全局
(2)区别:aggregateByKey操作<key,value>
(3)测试数据:

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)

每个分区中的元素(key,value)

def func3(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
iter.toList.map(x => "[partID:" + index + ", val: " + x + "]").iterator
}    


[partID:0, val: (cat,2)], [partID:0, val: (cat,5)], [partID:0, val: (mouse,4)],
[partID:1, val: (cat,12)], [partID:1, val: (dog,12)], [partID:1, val: (mouse,2)]

(4)把每个笼子中,每种动物最多的个数进行求和

pairRDD.aggregateByKey(0)(math.max(_,_),_+_).collect

 

 


4、coalesce和repartition

(*)都是将RDD中的分区进行重分区
(*)区别:coalesce 默认:不会进行shuffle(false)
        repartition 会进行shuffle
                    
(*)举例:
    创建一个RDD
    val rdd4 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),2)
             
    进行重分区
     val rdd5 = rdd4.repartition(3)
     val rdd6 = rdd4.coalesce(3,false)  ---> 分区的长度: 2
     val rdd6 = rdd4.coalesce(3,true)  ---> 分区的长度: 2

5、其他高级算子:参考文档
http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

以上是关于大数据笔记(二十九)——RDD简介特性及常用算子的主要内容,如果未能解决你的问题,请参考以下文章

2021年大数据Hadoop(二十九):​​​​​​​关于YARN常用参数设置

2021年大数据常用语言Scala(二十九):scala面向对象 单例对象

QT开发(二十九)——QT常用类

Java学习笔记之二十九Java中的"equals"和"=="的用法及区别

学习笔记第二十九节课

数学建模学习笔记(二十九)BP神经网络使用详例