Spark系列二：SparkCore的RDD算子Transformation和Action详解_rdd上支持的transformation和action算子

目录

概述

RDD 与 普通的集合有哪些区别

RDD的属性

创建RDD方式

RDD的算子  

RDD的Transformation的特点

常用的Transformation

aggregateByKey

​filterByRange

flatMapValues

foldByKey

combineByKey​

常用的Action

aggregate

collectAsMap

collect执行过程

 foreachPartition

countByKey 

 cache

checkpoint

总结

---

概述

1. RDD是一个基本的抽象
2. 操作RDD就像操作一个本地集合一样，不需要关心底层人物调度细节，降低了编程的复杂度。
3. RDD可以认为是一个代理
4.  RDD-（弹性分布式数据集）不存储真正要计算的数据，而是记录了RDD的转换关系（调用了什么方法，传入什么函数）
5. 对RDD操作，他会在Driver端转换成Task，下发到Executor计算分散在多台集群上的数据

RDD 与 普通的集合有哪些区别

RDD里面记入的是描述信息（从哪里读数据、以后对数据如何进行计算）、RDD的方法分为两类Transformation（lazy）、Action（生成Task，并发送到Executor中执行）
Scala存储的是真正要计算的数据、执行方法后立即回返回结果

RDD的属性

1. A list of partitions  （一系列分区，分区有编号，有顺序的），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
2. A function for computing each split  （每一个切片都会有一个函数作业在上面用于对数据进行处理），一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。
3. A list of dependencies on other RDDs  （RDD和RDD之间存在依赖关系）。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。
4. Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)，（可选，key value类型的RDD才有RDD[(K,V)]）如果是kv类型的RDD，会一个分区器，默认是hash-partitioned。即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
5. Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)，（可选，如果是从HDFS中读取数据，会得到数据的最优位置（向Namenode请求元数据））。一个列表，存储每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

创建RDD方式

        1.通过外部的存储系统创建RDD（读取HDFS文件）

            val rdd2 = sc.textFile("hdfs://lj01:9000/words.txt")

        2.将Driver的Scala集合通过并行化的方式编程RDD（试验、测验）

            val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8)）//由一个已经存在的Scala集合创建。
            val rdd1 = sc.makeRDD(List(1,2,3,4,5,6,7,8))，//makeRDD 底层调用的是parallelize

        3.调用一个已经存在了的RDD的Transformation，会生成一个新的RDD 

RDD的算子  

     RDD的算子分为两类，一类是Transformation（lazy），一类是Action（触发任务执行）

RDD的Transformation的特点

        1.lazy：不会直接计算结果，只是记住这些应用到一个文件上的转换动作，只有要求返回结果给Driver时，才会真正运行。
        2.生成新的RDD

常用的Transformation

aggregateByKey

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)
def func2(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
      iter.map(x => "[partID:" +  index + ", val: " + x + "]")
}
pairRDD.mapPartitionsWithIndex(func2).collect
pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect //(dog,12),(cat,19),(mouse,6)
pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect //(dog,112),(cat,219),(mouse,206)  它的初始值只在局部进行加，最后合并时不加初始值

filterByRange

val rdd1 = sc.parallelize(List(("e", 5), ("c", 3), ("d", 4), ("c", 2), ("a", 1)))
val rdd2 = rdd1.filterByRange("b", "d")
rdd2.colllect   //Array((c,3),(d,4),(c,2))  按照指定范围进行过滤,b到d,包含b和d

flatMapValues

val a = sc.parallelize(List(("a", "1 2"), ("b", "3 4")))
rdd3.flatMapValues(_.split(" "))  // Array((a,1),(a,2),(b,3),(b,4))

foldByKey

val rdd1 = sc.parallelize(List("dog", "wolf", "cat", "bear"), 2)
 
val rdd2 = rdd1.map(x => (x.length, x)) //Array((3,dog), (4,wolf), (3,cat), (4,bear))
 
rdd2.foldByKey("")(_+_).collect  //Array((3,dogcat), (4,wolfbear))
rdd2.aggregateByKey(" ")(_+_, _+_).collect  //Array((3,dogcat), (4,wolfbear))
rdd2.reduceByKey(_+_).collect  Array((3,dogcat), (4,wolfbear))
//以上三个方法底层都是调用"combineByKeywithClassTag"，都实现先局部聚合，在全局聚合

combineByKey

val rdd1 = sc.textFile("hdfs://lj01:9000/wc").flatMap(_.split(" ")).map((_, 1))
val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd2.collect
 
val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd3.collect
 
 
val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
val rdd6 = rdd5.zip(rdd4)
val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)

常用的Action

aggregate

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func1).collect
rdd1.aggregate(0)(math.max(_, _), _ + _)   // 18
rdd1.aggregate(5)(math.max(_, _), _ + _)   // 25
 
 
val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
      iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
rdd2.aggregate("")(_ + _, _ + _)   // abcdef 或者 defabc 两个task，并行计算
rdd2.aggregate("=")(_ + _, _ + _)   // ==abc=def 或者==def=abc
 
val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)  // 42 或者24  2和4进行拼接，并行执行，哪个先，哪个在前
 
val rdd4 = sc.parallelize(List("12","23","345",""),2)
rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)  // 10或者01 ，空串长度是0，1是前面哪个分区，得到0.toString,得到0,0的长度是1
 
val rdd5 = sc.parallelize(List("12","23","","345"),2)
rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)   // 11

collectAsMap

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap //Map(b->2,a->1)  最终结果弄成Map收集回来到Driver,这样不好
 
rdd.mapvalues(_*100).collectAsMap   //Map(b->200,a->100) 阻塞方法，一定要等结果计算返回。

collect执行过程：

 foreachPartition

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
rdd1.foreachPartition(x => println(x.reduce(_ + _)))

foreachPartition 一个分区一个分区  适合去数据库写数据

foreach   一条一条  具体结果是在excute的log文件查看

countByKey 

val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
rdd1.countByKey // Map(a->1,b->2,c->2)  只看出现次数，与里面数值无关
rdd1.countByValue

 cache

RDD相关的持久化和缓存，是Spark最重要的特征之一。可以说，缓存是Spark构建迭代式算法和快速交互式查询的关键。

RDD通过persist方法或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。

查看源码发现cache最终也是调用了persist方法

cache条件：
    1.要求的计算速度快
    2.集群的资源要足够大
    3.重要：cache的数据会多次的触发Action
    4.先进行过滤，然后将缩小范围的数据在cache到内存

默认的存储级别都是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的。

    //第一个参数，放到磁盘（速度快）
    //第二个参数，放到内存 （安全 ）
    //第三个参数，磁盘中的数据，不是以java对象的方式保存，序列化（默认的序列化）
    //第四个参数，内存中的数据，以java对象的方式保存，不序列化

    // 2,保存两份，在不同机器上 
    val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)

    OFF_HEAP:堆外内存

       例如：

              achyon，开源分布式内存存储系统

      改名 Alluxio - Open Source Memory Speed Virtual Distributed Storage

checkpoint

使用checkpoint场景：
    1.迭代计算，要求保证数据安全
    2.对速度要求不高（跟cache到内存进行对比）
    3.将中间结果保存到hdfs
 案列：

sc.setCheckpointDir("hdfs://lj01:9000/ck") //设置checkpoint目录（分布式文件系统的目录hdfs目录）
val rdd = sc.textFile("hdfs://lj01:9000/wc").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)    //经过复杂进行，得到中间结果
rdd.checkpoint   //将中间结果checkpoint到指定的hdfs目录，是个标记，之前的父依赖关系就不用了
rdd.isCheckpointed
rdd.count   //后续的计算，就可以使用前面ck的数据了，做了两件事：写入HDFS，计算结果，所以可以在checkpoint之前先写入内存（避免相同的数据读两次），那么之后直接将内存写到HDFS，然后释放内存的数据
rdd.isCheckpointed
rdd.getCheckpointFile   

就做了cache又做了checkpoint，那么

在if判断在cache，内存找数据

在if判断是否checkpoint，HDFS找数据

在找父依赖看数据

总结

RDD是spark中的一个最基本的抽象，代表着一个不可变、可分区、可以并行计算的分布式数据集
RDD有5个特点
    1.一系列分区
    2.会有一个函数作用在每个切片上
    3.RDD和RDD之间存在依赖关系
    4（可选）如果是RDD中装的是KV类型的，那么Shuffle时会有一个分区器。默认是HashPartitioner
    5（可选）如果只从HDFS中读取数据，会感知数据则位置，将Executor启动在数据所在的机器上

Spark的设计就是基于这个抽象的数据集（RDD），你操作RDD这个抽象的数据集，就像操作一个本地集合一样，Spark包底层的细节都隐藏起来的（任务调度、Task执行，任务失败重试等待），开发者使用起来更加方便简洁

操作RDD，其实是对每个分区进行操作，分区会生成Task，Task会调度Executor上执行相关的计算逻辑，进而对数据进操作

---

如果您看到这了，请您动动小手点个赞或关注一下，鼓励一下我，谢谢您