第二章 Spark RDD以及编程接口

小鱼儿 2023-07-05 13:23 50阅读 0赞

# 第二章 Spark RDD以及编程接口 #

--------------------

## 目录 ##

1.  Spark程序"Hello World"
2.  Spark RDD
3.  创建操作
4.  转换操作
5.  控制操作
6.  行动操作

注：学习《Spark大数据处理技术》笔记

--------------------

### 1. Spark程序"Hello World" ###

#### 1. 概述 ####

1.  计算存储在HDFS的Log文件中出现字符串"Hello World"的行数

##### 2. 代码实现 #####

![在这里插入图片描述][20200215134120183.png]

#### 3. 行解 ####

###### 第一行 ######

1.  对于所有的Spark程序而言，要进行任何操作，首先要创建一个Spark上下文，在创建上下文的过程中，程序会向集群申请资源以及构建相应的运行环境
2.  需要传入四个变量  
    1. Spark程序运行的集群地址，如"spark://localhost:7077"  
    2. Spark程序的标识  
    3. 指明Spark程序安装的路径  
    4. Spark程序的jar包路径

###### 第二行 ######

1.  通过sc变量，利用textFile接口从HDFS文件系统读入Log文件，返回一个变量file

###### 第三行 ######

1.  对file变量进行过滤操作，传入的参数是一个function对象，function的原型p:(A)=>Boolean，对于file中的每一行字符串判断是否含有"Hello World"，生成新的变量filterRDD

###### 第四行 ######

1.  对filterRDD进行cache操作，以便后续重用filterRDD这个变量

###### 第五行 ######

1.  对filterRDD进行count计数，最后返回包含"Hello World"字符串的文本行数

#### 4. 重要概念 ####

1.  弹性分布式数据集RDD(Resilient Distributed DataSets)  
    file和filterRDD变量都是RDD
2.  创建操作(creation operation)  
    RDD的初始创建都是由SparkContext来负责的，将内存中的集合或者外部文件系统作为输入源
3.  转换操作(transformation operation)  
    将一个RDD通过一定的操作变换成另一个RDD，比如file这个RDD通过一个filter操作变换成filterRDD，所以filter就是一个转换操作
4.  控制操作(control operation)  
    对RDD进行持久化，可以让RDD保存在磁盘或者内存中，以便后续重复使用。比如cache接口默认将filterRDD缓存在内存中
5.  行动操作(action operation)  
    由于Spark是惰性计算(lazy computing)的，所以对于任何RDD进行行动操作，都会触发Spark作业的运行，从而产生最终的结果。例如对filterRDD进行count操作就是一个行动操作。行动操作分为两类：
    
    1.  一类的操作结果变成Scala集合或者标量
    2.  另一类就是将RDD保持到外部文件或者数据库系统中

#### 5. RDD与操作之间的关系 ####

1.  经过输入操作(创建操作)、转换操作、控制操作、输出操作(行动操作)来完成一个任务。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70]

--------------------

### 2. Spark RDD ###

#### 1. 是什么？ ####

1.  RDD是弹性分布式数据集，即一个RDD代表一个被分区的只读数据集

#### 2. 如何生成？ ####

1.  内部集合和外部存储系统
2.  通过转换操作来自其他RDD，比如map、filter、join等等

#### 3. 特点？ ####

###### 1. 容错处理 ######

1.  RDD的接口只支持粗粒度的操作（即一个操作会被应用在RDD的所有数据上），所以只要通过记录这些作用在RDD上的转换操作，来构建RDD的继承关系(lineage)，就可以有效进行容错处理，而不需要将实际的RDD数据进行记录拷贝
2.  也就是在一个Spark程序中，我们用到的每一个RDD，在丢失或者操作失败后都是可以重建的

###### 2. 持久化和分区 ######

1.  开发者可以指明需要重用哪些RDD，然后选择一种存储策略（如 in-memory storage）将它们保存起来
2.  开发者可以让RDD根据记录中的健值在集群的机器之间重新分区，这对于RDD位置优化非常有用，如：将要进行join操作的两个数据集以同样的方式进行哈希分区

#### 4. 一个分区的、高效容错的而且能够持久化的分布式数据集需要包含五个接口 ####

###### 1. RDD分区（partitions） ######

1.  RDD具备分区的属性，对于一个RDD而言，分区的多少涉及到对这个RDD进行并行计算的粒度，每一个RDD分区的计算操作都在一个单独的任务中执行。
2.  用户可以自行指定分区，如果没有指定，则使用系统默认的值（这个程序所分配到的资源的CPU核的个数）
3.  可以利用RDD的成员变量partitions所返回的partition数组的大小查询一个RDD被划分的分区数

###### 2. RDD优先位置（preferredLocations） ######

1.  RDD优先位置属性与Spark中的调度相关，返回的是此RDD的每个partition所存储的位置，按照“移动数据不如移动计算”的理念，在Spark进行任务调度的时候，尽可能将任务分配到数据块所存储的位置。
2.  以从Hadoop中读取数据生成RDD为例，perferredLocations返回每一个数据块的机器名或者IP地址，如果每一块数据是多份存储的，就返回多个机器地址

###### 3. RDD依赖关系（dependencies） ######

1.  由于RDD是粗粒度的操作数据集，每一个转换操作都会生成一个新的RDD，所以RDD之间就会形成类似流水线一样的前后依赖关系
2.  Spark中存在两种类型的依赖
    
    1.  窄依赖  
        每一个父RDD的分区最多被子RDD的一个分区所使用
    2.  宽依赖  
        多个子RDD的分区依赖于同一个父RDD的分区
    3.  区别
        
        1.  执行
            
            1.  窄依赖可以在集群的一个节点上如流水线一般地执行，可以计算所有父RDD的分区
            2.  宽依赖需要取得父RDD的所有分区上的数据进行计算，将会执行类似于MapReduce一样的Shuffle操作
        2.  节点恢复
            
            1.  对于窄依赖来说，节点计算失败后的恢复更加有效，只要重新计算对应的父RDD的分区，而且可以在其他节点上并行地计算
            2.  在宽依赖的继承关系中，一个节点的失败将会导致其父RDD的多个分区重新计算，代价是非常高的

###### 4. RDD分区计算（compute） ######

1.  对于Spark中每个RDD的计算都是以partition（分区）为单位的，而且RDD中的compute函数都是在对迭代器进行复合，不需要保存每次计算的结果

###### 5. RDD分区函数（partitioner） ######

1.  partitioner就是RDD分区函数，Spark实现两种类型的分区函数，HashPartitioner（哈希分区）和RanagePatitioner（区域分区），且partitioner这个属性只存在于（K,V）类型的RDD中，非（K,V）类型的partitioner的值就是None
2.  partition函数既决定了RDD本身的分区数量，也可作为其父RDD Shuffle输出中每个分区进行数据切割的依据

--------------------

### 3. 创建操作 ###

#### 1. 集合创建操作 ####

1.  RDD的形成可以由内部集合类型来生成。Spark中提供了parallelize 和 makeRDD两类函数来实现集合生成RDD
2.  两个函数接口功能类似，不同的是makeRDD还提供了一个可以指定每一个分区preferredLocations 参数的实现版本

#### 2. 存储创建操作 ####

1.  Spark的整个生态系统与Hadoop是完全兼容的，所以对于Hadoop所支持的文件类型或者数据库类型，为了兼容hadoop所有版本，提供了两套创建操作接口。对于外部存储创建操作而言，hadoopRDD和newHadoopRDD是最为抽象的两个函数接口，主要包括以下四个参数
    
    1.  输入格式  
        指定数据输入的类型，如TextInputFormat
    2.  键类型  
        指定\[K,V\]健值中K的类型
    3.  值类型  
        指定\[K,V\]健值中V的类型
    4.  分区值  
        指定由外部存储生成的RDD的partition数量的最小值，如果没有指定，系统会使用默认值defaultMinSplits

--------------------

### 4. 转换操作 ###

1.  mapU:ClassTag: RDD\[U\]  
    map 函数将RDD中类型为 T 的元素，一对一地映射为类型为 U 的元素。
2.  distinct(): RDD\[T\]  
    distinct 函数返回RDD中所有不一样的元素
3.  flatMapU:ClassTag:RDD  
    flatMap 函数将RDD中的每一个元素进行一对多转换
4.  repartition(numPartitions:Int):RDD\[T\]  
    repartition 只是 coalesce 接口中shuffle 为 ture 的简易实现
5.  coalesce(numPartitions:Int,shuffle:Boolean=false):RDD\[T\]  
    假设RDD有N个分区，需要重新划分成M个分区
    
    1.  如果N<M，一般情况下N个分区有数据分布不均的状况，利用HashPartitioner函数将数据重新分区为M个，这时需要将shuffle参数设置为true
    2.  如果N>M且N和M相差不多（比如N是1000，M是100），那么就可以将N个分区中的若干个分区合并成一个新的分区，最终合并成M个分区，这时可以将shuffle设置为false（shuffle在false情况下，设置M>N，coalesce是不起作用的），不进行shuffle过程，父RDD和子RDD是窄依赖关系
    3.  如果N>M且N和M差距悬殊（比如N是1000，M是1），这个时候如果把shuffle参数设置为false，由于父子RDD是窄依赖，它们同处在一个stage中，就可能会造成spark程序运行的并行度不够，从而影响性能。比如在M为1时，由于只有一个分区，所以只会有一个任务在运行，为了使coalesce之前的操作有更好的并行度，可以将shuffle设置为true
6.  randomSplit(weights: Array\[Double\],seed: Long=System.nanoTime): Array\]\[RDD\[T\]\]  
    randomSplit函数是根据weights权重将一个RDD切分成多个RDD
7.  glom: RDD\[Array\[T\]\]  
    而glom函数是将RDD中每一个分区中类型T的元素转换成数组Array\[T\]，这样每一个分区就只有一个数组元素
8.  union(other: RDD\[T\]): RDD\[T\]  
    这些是针对RDD的集合操作，union操作将两个RDD的数据进行合并，返回两个RDD的并集（包含两个RDD中相同的元素，不会去重）
9.  intersection(other: RDD\[T\], partitioner: Partitoner)  
    intersection操作返回两个RDD集合的交集，且交集中不会包含相同的元素
10. subtract(other: RDD\[T\]): RDD\[T\]  
    如果subtract所针对的两个集合A和B，即操作是 val result=A.subtract(B)，那么result中将会包含在A中出现且B中不出现的元素。intersection和subtract一般情况都会有shuffle的过程
11. subtract(other:RDD\[T\], p:Partitioner): RDD\[T\]
12. mapPartitions\[U: ClassTag\](f: Iterator\[T\]=>Iterator\[U\], parservesPartitioning: Boolean = false): RDD\[U\]  
    mapPartitions与map转换操作类似，只不过映射函数的输入参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器，那么已经有了map为什么还要mapPartitions函数呢？如果在映射的过程中需要频繁创建额外的对象，map就显得不高效了。例如，将RDD中的所有数据通过JDBC连接写入数据库中，如果使用map函数可能需要为每一个元素都创建一个connection，这样开销是很大的，如果利用mapPartitions接口，可以针对每一个分区创建一个connection。
13. mapPartitonsWithIndex\[U:ClassTag\](f: (Int,Iterator\[T\])=>Iterator\[U\], preservesPartitioning: Boolean=false): RDD\[U\]  
    mapPartitionsWithIndex和mapPartitions功能类似，只是输入参数多了一个分区的ID。
14. zipU:ClassTag:RDD\[T, U\]  
    zip函数的功能是将两个RDD组合成Key/Value（健/值）形式的RDD，这里默认两个RDD的partition数量以及元素数量都相同，否则相同系统将会抛出异常。
15. zipPartitions\[B: ClassTag, V:ClassTag\](rdd2:RDD\[B\], preservesPartitioning:Boolean)(f:(Iterator\[T\],Iterator\[B\])=>Iterator\[V\]:RDD\[V\])  
    zipPartitions是将多个RDD安装partition组合成为新的RDD，zipPartitions需要相互组合的RDD具有相同的分区数，但是对于每个分区中的元素数量是没有要求的
    
    未完待续…

### 5. 控制操作 ###

1.  cache(): RDD\[T\]
2.  persist(): RDD\[T\]
3.  persist(level:StorageLevel): RDD\[T\]
    
    1.  在Spark中对RDD进行持久化操作是一项非常重要的功能，可以将RDD持久化在不同层次的存储介质中，以便后续的操作能够重复使用，这对iterative和interactive的应用来说会极大地提高性能
4.  checkpoint
    
    1.  checkpoint接口是将RDD持久化在HDFS上，其与persist（如果也持久化在磁盘上）的一个区别是checkpoint将会切断此RDD之前的依赖关系，而persist接口依然保留着RDD的依赖关系
    2.  checkpoint主要作用
        
        1.  如果一个Spark程序会长时间驻留运行（如sparkStream一般会7x24小时运行），过长的依赖将会占用很多系统资源，那么定期将RDD进行checkpoint操作，能够有效节省系统资源
        2.  维护过长的依赖关系还会出现一个问题，如果Spark在运行过程中出现节点失败的情况，那么RDD进行容错重算成本会非常高

### 6. 行动操作 ###

#### 1. 概述 ####

1.  行动操作是和转换操作相对应的一种对RDD的操作类型，在Spark 程序中，每调用一次行动操作，都会触发一次Spark的调度并返回相应的结果。
2.  行动操作分为两种类型
    
    1.  行动操作将标量或者集合返回给Spark客户端程序，比如返回RDD中数据集的数量或者是返回RDD中的一部分符合条件的数据。
    2.  行动操作将RDD直接保存到外部文件系统或者数据库中，比如将RDD保存到HDFS文件系统中

#### 2. 集合标量行动操作 ####

1.  first：返回RDD中第一个元素
2.  count：返回RDD中元素的个数
3.  reduce(f:(T, T)=>T)：对RDD中的元素进行二元计算，返回计算结果
4.  collect()/toArray()：以集合形式返回RDD的元素
5.  take(num: Int)：将RDD作为集合，返回集合中\[0, num-1\]下标的元素
6.  top(num: Int)：按照默认的或者是指定的排序规则，返回前num个元素
7.  takeOrdered(num: Int)：以与top相反的排序规则，返回前num个元素
8.  aggregate\[U\](zeroValue: U)(seqOp: (U, T)=>U, combOp(U, U)=>U)  
    aggregate行动操作中主要需要提供两个函数，一个是seqOp函数，其将RDD（RDD中的每个元素的类型是T）中的每一个分区的数据聚合成类型为U的值。另一个函数combOp将各个分区聚合起来的值合并在一起得到最终类型U的返回值。这里的RDD元素的类型T和返回值的类型U可以为同一个类型。
9.  fold(zeroValue: T)(op: (T, T)=>T)  
    fold是aggregate的便利接口，其中，op操作既是seqOp操作也是combOp操作，且最终的返回类型也是T，即与RDD中每一个元素的类型是一样的
10. lookup(key: K): Seq\[V\]  
    lookup是针对(K, V)类型RDD的行动操作，对于给定的健值，返回与此健值相对应的所有值

#### 3. 存储行动操作 ####

###### 1. 概述 ######

1.  对于RDD最后的归宿除了返回为集合和标量，也可以将RDD存储到外部文件系统或者数据库中，Spark系统和Hadoop是完全兼容的，所以对MapReduce所支持的读写文件或者数据库类型，也同样支持。此外，由于Hadoop的API有新旧两个版本，所以Spark为了能兼容Hadoop所有的版本，也提供了两套API

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70 1]

1.  这里一共列出七个将RDD存储到外部介质的旧版API，前六个API都是saveAsHadoopDataset这个API的简易实现版本，仅仅支持将RDD存储到HDFS中，而saveAsHadoopDataset的参数类型是jobConf，所以其不仅能将RDD存储到HDFS上，也可以将RDD存储到其它数据库中，如Hbase，MangoDB等
2.  将RDD保存到HDFS中通常情况需要关注或者设置五个参数，即文件保存的路径、key值的class类型，Value值的class类型，RDD的输出格式（outputFormat，如TextOutputFormat），以及最后一个相关的参数codec(如DefaultCodec、Gzip等)

###### 2. 在Spark中针对新版Hadoop API提供了三个行动操作函数 ######

![在这里插入图片描述][20200215142355140.png]

1.  与旧版的Hadoop API的接口使用方法类似，前两个API支持将RDD保存到HDFS中，而saveAsNewAPIHadoopDataset则支持所有MapReduce兼容的输入输出类型。主要不同点是由于Hadoop API的改变，输入参数的类型发生了变化。

[20200215134120183.png]: https://img-blog.csdnimg.cn/20200215134120183.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200215134922440.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200215142247661.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA==,size_16,color_FFFFFF,t_70
[20200215142355140.png]: https://img-blog.csdnimg.cn/20200215142355140.png