RDDs基本操作RDDs特性KeyValue对RDDs

Posted 2020-09-30

摘要：RDD是Spark中极为重要的数据抽象，这里总结RDD的概念，基本操作Transformation(转换)与Action，RDDs的特性，KeyValue对RDDs的Transformation(转换)。

1.RDDs是什么

Resilient distributed datasets(弹性分布式数据集) 。RDDs并行的分布在整个集群中,是Spark分发数据和计算的基础抽象类,一个RDD是一个不可改变的分布式集合对象，Spark中，所有的计算都是通过RDDs的创建，转换操作完成的，一个RDD内部由许多partitions(分片)组成。

分片：每个分片包括一部分数据，partitions可在集群不同节点上计算；分片是Spark并行处理的单元，Spark顺序的，并行的处理分片。

2.RDDs的创建

(1)   把一个存在的集合传给SparkContext的parallelize()方法，测试用
　　val rdd=sc.parallelize(Array(1,2,3,4),4)
　　第一个参数：待并行化处理的集合；第二个参数：分片个数

(2)  加载外部数据集
　　val rddText=sc.textFile("hellospark.txt")

3.RDD基本操作之Transformation(转换)

从之前的RDD构建一个新的RDD,像map()和filter()

(1)逐元素Transformation
map():   接收函数，把函数应用到RDD的每一个元素，返回新RDD。

　　val lines=sc.parallelize(Array("hello","spark","hello","world","!"))
　　val lines2=lines.map(word=>(word,1))
　　lines2.foreach(println)
　　//结果：
　　(hello,1)
　　(spark,1)
　　(hello,1)
　　(world,1)
　　(!,1)

filter():   接收函数，返回只包含满足filter()函数的元素的新RDD。 

　　val lines=sc.parallelize(Array("hello","spark","hello","world","!"))
　　val lines3=lines.filter(word=>word.contains("hello"))
　　lines3.foreach(println)
　　//结果：　　　
　　hello
　　hello

flatMap(): 对每个输入元素，输出多个输出元素。flat压扁的意思，将RDD中元素压扁后返回一个新的RDD。

    val inputs=sc.textFile("/home/lucy/hellospark.txt")
    val lines=inputs.flatMap(line=>line.split(" "))
    lines.foreach(println)
    //结果
    hello
    spark
    hello
    world
    hello
    !
    //文件内容/home/lucy/hellospark.txt 
    hello spark
    hello world
    hello !

(2)集合运算

RDDs支持数学集合的计算，例如并集，交集计算

    val rdd1=sc.parallelize(Array("red","red","blue","black","white"))
    val rdd2=sc.parallelize(Array("red","grey","yellow"))

    //去重：
    val rdd_distinct=rdd1.distinct()
    //去重结果：
    white
    blue
    red
    black

    //并集：
    val rdd_union=rdd1.union(rdd2)
    //并集结果：
    red
    blue
    black
    white
    red
    grey
    yellow

    //交集：
    val rdd_inter=rdd1.intersection(rdd2)
    //交集结果：
    red

    //包含：
    val rdd_sub=rdd1.subtract(rdd2)
    //包含结果：
    blue
    white
    black

4.RDD基本操作之Action

在RDD上计算出来一个结果。把结果返回给driver program或保存在文件系统，count(),save。

函数名 　　　　　　　　　　　　　功能 　　　　　　　　　　　　          例子 　　                                    结果
collect()　　 　　 　　　　　 返回RDD的所有元素　　　　　　　　　　　rdd.collect()　　　　　　 　　 {1,2,3,3}
count() 　　 　　 　　　        计数　　　　　　　　　　　　　　　　　    rdd.count()　　　　　　              4
countByValue() 　　 　　　   返回一个map表示唯一元素出现的个数 　　   rdd.countByValue()　　            {(1,1),(2,1),(3,2)}
take(num) 　　 　　           　返回几个元素 　　　　　　　　　　 　　     rdd.take(2) 　　 　　                 {1,2}
top(num) 　　 　　　             返回前几个元素 　　　　　　　　　　         rdd.top(2)　　　　　　               {3,3}
takeOrdered                            返回基于提供的排序算法的前几个元素          rdd.takeOrdered(2)(myOrdering)　{3,3}
(num)(ordering)
takeSample                             取样例　　　　　　　　　　　　　　          rdd.takeSample(false,1) 　　       不确定
(withReplacement,num,[seed]) 　　　　　　　　　　　　　　　　　　　　　 　　　　
reduce(func) 　　 　　            合并RDD中元素 　　　　　　　　　　         rdd.reduce((x,y)=>x+y) 　　　　 9
fold(zero)(func) 　　               与reduce()相似提供zero value　　　　　       rdd.fold(0)((x,y)=>x+y) 　             9
foreach(func) 　　                   对RDD的每个元素作用函数，什么也不返回   rdd.foreach(func) 　　　　　       无

5.RDDs的特性

1.血统关系图：
        Spark维护着RDDs之间的依赖关系和创建关系，叫做血统关系图，Spark使用血统关系图来计算每个RDD的需求和恢复丢失的数据
2.延迟计算
       Spark对RDDs的计算是，他们第一次使用action操作的时候。Spark内部记录metadata表明transformations操作已经被响应了。加载数据也是延迟计算，数据只有在必要的时候，才会被加载进去。
3.RDD持久化

       Spark最重要的一个功能是它可以通过各种操作持久化（或者缓存）一个集合到内存中。当持久化一个RDD的时候，每一个节点都将参与计算的所有分区数据存储到内存中，并且这些数据可以被这个集合（以及这个集合衍生的其他集合）的动作（action）重复利用。这个能力使后续的动作速度更快（通常快10倍以上）。对迭代算法和快速的交互使用来说，缓存是一个关键的工具。

    可以通过persist()或者cache()方法持久化一个rdd。首先，在action中计算得到rdd；然后，将其保存在每个节点的内存中。Spark的缓存是一个容错的技术-如果RDD的任何一个分区丢失，它可以通过原有的转换（transformations）操作自动的重复计算并且创建出这个分区。

    此外，可以利用不同的存储级别存储每一个被持久化的RDD。例如，它允许我们持久化集合到磁盘上、将集合作为序列化的Java对象持久化到内存中、在节点间复制集合或者存储集合到Tachyon中。我们可以通过传递一个StorageLevel对象给persist()方法设置这些存储级别。cache()方法使用了默认的存储级别—StorageLevel.MEMORY_ONLY。完整的存储级别如下：

Storage Level	Meaning
DISK_ONLY,&_2	仅仅将RDD分区存储到磁盘中
MEMORY_ONLY,&_2	将RDD作为非序列化的Java对象存储在jvm中。如果内存装不下原始文件那么大的数据，一些分区将不会被缓存，从而在每次需要这些分区时都需重新计算它们。这是系统默认的存储级别。
MEMORY_ONLY_SER,&_2	将RDD作为序列化的Java对象存储（每个分区一个byte数组）。这种方式比非序列化方式更节省空间，特别是用到快速的序列化工具时，但是会更耗费cpu资源—密集的读操作。
MEMORY_AND_DISK,&_2	将RDD作为非序列化的Java对象存储在jvm中。如果内存装不下原始文件那么大的数据，将这些不适合存在内存中的分区存储在磁盘中，每次需要时读出它们。
MEMORY_AND_DISK_SER,&_2	和MEMORY_ONLY_SER类似，但不是在每次需要时重复计算这些不适合存储到内存中的分区，而是将这些分区存储到磁盘中。
OFF_HEAP (experimental)	以序列化的格式存储RDD到Tachyon中。相对于MEMORY_ONLY_SER，OFF_HEAP减少了垃圾回收的花费，允许更小的执行者共享内存池。这使其在拥有大量内存的环境下或者多并发应用程序的环境中具有更强的吸引力。

      

NOTE:在python中，存储的对象都是通过Pickle库序列化了的，所以是否选择序列化等级并不重要。

Spark也会自动持久化一些shuffle操作（如reduceByKey）中的中间数据，即使用户没有调用persist方法。这样的好处是避免了在shuffle出错情况下，需要重复计算整个输入。

如何选择存储级别？

1.如果你的RDD适合默认的存储级别（MEMORY_ONLY），就选择默认的存储级别。因为这是cpu利用率最高的选项，会使RDD上的操作尽可能的快。

2.如果不适合用默认的级别，选择MEMORY_ONLY_SER。选择一个更快的序列化库提高对象的空间使用率，但是仍能够相当快的访问。

3.除非函数计算RDD的花费较大或者它们需要过滤大量的数据，不要将RDD存储到磁盘上，否则，重复计算一个分区就会和重磁盘上读取数据一样慢。

4.如果你希望更快的错误恢复，可以利用重复(replicated)存储级别。所有的存储级别都可以通过重复计算丢失的数据来支持完整的容错，但是重复的数据能够使你在RDD上继续运行任务，而不需要重复计算丢失的数据。

Spark自动的监控每个节点缓存的使用情况，利用最近最少使用原则删除老旧的数据。如果你想手动的删除RDD，可以使用RDD.unpersist()方法 

6.KeyValue对RDDs

创建KeyValue对RDDs：

val rdd3=sc.parallelize(Array((1,2),(3,4),(3,6)))

KeyValue对RDDs的Transformation(转换):

(1)reduceByKey(func)    把相同key的结合 

　　val rdd4=rdd3.reduceByKey((x,y)=>x+y)
　　//结果
　　(1,2)
　　(3,10)

(2)groupByKey      把相同的key的values分组

　　val rdd5=rdd3.groupByKey()
　　//结果
　　(1,CompactBuffer(2))
　　(3,CompactBuffer(4, 6)) 

(3)mapValues()    函数作用于pairRDD的每个元素，key不变

　　val rdd6=rdd3.mapValues(x=>x+1)
　　//结果
　　(1,3)
　　(3,5)
　　(3,7)

(4)keys/values
　　rdd3.keys.foreach(println)
　　1
　　3
　　3

　　rdd3.values.foreach(println)
　　2
　　4
　　6

(5)sortByKey
　　val rdd7=rdd3.sortByKey()
　　//结果
　　(1,2)
　　(3,4)
　　(3,6)

(6)combineByKey():     (createCombiner,mergeValue,mergeCombiners,partitioner)

　　最常用的基于key的聚合函数，返回的类型可以与输入类型不一样。许多基于key的聚合函数都用到了它，像groupByKey()

　　原理：遍历partition中的元素，元素的key，要么之前见过的，要么不是。如果是新元素，使用我们提供的createCombiner()函数，如果是这个partition中已经存在的key,就会使用mergeValue()函数，合计每个partition的结果的时候，使用mergeCombiner()函数

　　例子：求平均值

　　val score=sc.parallelize(Array(("Tom",80.0),("Tom",90.0),("Tom",85.0),("Ben",85.0),("Ben",92.0),("Ben",90.0)))
　　val score2=score.combineByKey(score=>(1,score),(c1:(Int,Double),newScore)=>(c1._1+1,c1._2+newScore),(c1:(Int,Double),c2:(Int,Double))=>(c1._1+c2._1,c1._2+c2._2))
　　//结果
　　(Ben,(3,267.0))
　　(Tom,(3,255.0))

    val average=score2.map{case(name,(num,score))=>(name,score/num)}
　　//结果
　　(Ben,89.0)
　　(Tom,85.0)

7.RDD依赖

Spark中RDD的高效与DAG图有着莫大的关系，在DAG调度中需要对计算过程划分stage，而划分依据就是RDD之间的依赖关系。针对不同的转换函数，RDD之间的依赖关系分类窄依赖（narrow dependency）和宽依赖（wide dependency, 也称 shuffle dependency）.

窄依赖是指父RDD的每个分区只被子RDD的一个分区所使用；宽依赖是指父RDD的每个分区都可能被多个子RDD分区所使用：

宽依赖和窄依赖如下图所示：

这种划分有两个用处。首先，窄依赖支持在一个结点上管道化执行。例如基于一对一的关系，可以在filter之后执行map。其次，窄依赖支持更高效的故障还原。因为对于窄依赖，只有丢失的父RDD的分区需要重新计算。

对于宽依赖，一个结点的故障可能导致来自所有父RDD的分区丢失，因此就需要完全重新执行。因此对于宽依赖，Spark会在持有各个父分区的结点上，将中间数据持久化来简化故障还原，就像MapReduce会持久化map的输出一样。