Spark RDD数据操作函数以及转换函数一文详解运用与方法

前言：

配置JDK1.8

实验环境IDEA

scala版本为2.11.12

本地Window伪分布运行非集群实验

创建RDD

从内存中创建一个RDD有两种常用的方法。一种是转化Seq集合为RDD，另一种是从已有RDD转化为新的RDD。

SparkContext类中有两个方法：parallelize和makeRDD。

1.parallelize

parallelize有两个参数可以输入

（1）要转化的集合，必须是Seq集合。

（2）分区数，一般不设分区数，则默认为该Application分配到的资源的CPU数。

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

2.makeRDD

makeRDD有两种实现方法：一种跟parallelize完全一致；另一种接收的参数类型是Seq，生产的RDD中保存的是T的值（Seq[T,Seq[String])）。

val seq = Seq((1,Seq(1,2)),(2,Seq(2,3,4)))
val rdd =sc.makeRDD(seq)
rdd.collect().foreach(println(_))                     

(1,List(1, 2))

(2,List(2, 3, 4))

从外部存储创建RDD是指直接读取一个存放在文件系统的数据文件创建RDD，第一种创建RDD的方式常用于测试，这种方式才是用于实践操作的常用方法。

（1）从HDFS文件创建RDD

val test = sc.textFile("/user/root/test.txt")

（2）从Linux本地文件创建

确实差不多，在路径前面加上file：//表示从本地Linux文件系统读取。

1.Map转换数据

map是一种基础的RDD转换操作，用于将RDD中的每一个数据元素通过某种函数进行转换并返回新的RDD。

例：

val distData = List(1, 3, 45, 3, 76)
val sq_dist = distData.map(x => x * x)
print(sq_dist)

List(1, 9, 2025, 9, 5776)

2.SortBy()排序

sortBy()是对标准RDD进行排序的方法，在org.apache.spark.rdd.RDD类中实现：

/**
 * Return this RDD sorted by the given key function.
 */
def sortBy[K](
    f: (T) => K,
    ascending: Boolean = true,
    numPartitions: Int = this.partitions.size)
    (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T] =
  this.keyBy[K](f)
      .sortByKey(ascending, numPartitions)
      .values

第一个参数是函数f(x)=>_._._，左边是要被排序对象中的每一个元素，右边返回的值是元素中要进行排序的值。

第二个参数是ascending排序顺序，决定排序后RDD中的元素是升序还是降序，默认是ture

第三个参数是numPartitions，该参数决定排序后的RDD分区个数，默认排序后的分区个数和排序之前的个数相等。

例：

val data = List((5,3),(888,666),(777,65))
val sort_data=data.sortBy(x=>x._1)
print(sort_data)

List((5,3), (777,65), (888,666))

3.collect()查询

collect函数是一个行动操作，把RDD所有元素转换成数组并返回到Driver端，适用于小数据处理后的返回。

sq_data.collect

Array[(Int,Int)] = Array((7,6),(45,3),(1,3))

4.flatMap转换数据

faltMap的操作是将函数应用于RDD之中的每一个元素，将返回的迭代器中的所有元素构成新的RDD。

简单的来讲，使用faltmap就是先map然后flat迭代输出：

val test = List("How are you", "I am fine", "What about you")
print(test.flatMap(x => x.split(" ")))

List(How, are, you, I, am, fine, What, about, you)

5.take()查询指定数目的值

take(N)方法用于获取RDD的前N个元素，返回类型为数组。take与collect的原理相似，collect用于获取全部数据，take获取指定个数的数据。

val data = sc.parallelize(1 to 10)
data.take(5)

Array[Int] = Array(1,2,3,4,5)

6.union()合并多个RDD

union是一种转换操作，用于将两个RDD的元素合并成一个，不进行去重操作，而且两个RDD中每个元素中的值的个数和类型需要保持一直。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5)))
rdd1.union(rdd2).collect

((a,1),(b,2),(c,3),(a,1),(d,4),(e,5))

7.filter()进行过滤

filter是一种转换操作，用于过滤RDD中的元素。filter需要一个参数，参数是一个用于过滤的函数，该函数的返回值为Boolean类型，返回值为true的元素保留，返回值为false的元素过滤，最后结果是返回一个存储符合过滤条件的所有元素的新RDD。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
rdd1.filter(_._2>1).collect.foreach(println(_))

(b,2)

(c,3)

8.distinct()进行去重

distinct()是一个转换操作，用于RDD的数据去重，去除两个完全相同的元素，没有参数。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3),('a',1)))
rdd1.distinct().collect().foreach(println(_))

(b,2)

(c,3)

(a,1)

9.intersection()求出两个RDD的共同元素

intersection()方法用于求出两个RDD的共同元素，也就是找出两个RDD的交集，参数是另一个RDD，顺序先后与结果无关。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3),('a',1)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5)))
rdd1.intersection(rdd2).collect().foreach(println(_))

(a,1)

10.subtract()将相同元素去掉

subtract()的参数是一个RDD，用于将前一个RDD中在后一个RDD出现的元素删除，可以看作是求补集的操作，返回值为前一个RDD去除与后一个RDD相同的元素后的剩余值所组成的新的RDD，所以RDD的顺序会影响结果。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5),('b',2)))
rdd1.subtract(rdd2).collect().foreach(println(_))
rdd2.subtract(rdd1).collect().foreach(println(_))

(c,3)

(e,5)

(d,4)

11.cartesian()求两个RDD的笛卡尔积

笛卡尔积就是将两个集合的元素两两组合成一组，假设集合A有5个元素，集合B有10个元素，集合A的每个元素都会和集合B的每个元素组合成一组，结果会返回50个元素组合。

val rdd1 = sc.parallelize(List(1,2,3,4))
val rdd2 = sc.parallelize(List(1,2,3))
rdd1.cartesian(rdd2).collect().foreach(println(_))

(1,1)

(1,2)

(1,3)

(2,1)

(2,2)

(2,3)

(3,1)

(3,2)

(3,3)

(4,1)

(4,2)

(4,3)

键值对RDD

键值对RDD由一组组的键值对组成，这些RDD被称为PairRDD。PairRDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口。

val rdd= sc.parallelize(List("this is a test","hellow world ","come on "))
val words = rdd.map(x=>(x.split(" ")(0),x));
words.collect().foreach(println(_))

(this,this is a test)

(hellow,hellow world )
(come,come on )

转换操作Keys与Values

作为键值对类型的RDD，包含了键和值两部分。Spark提供了两种方法，分别获取键值对RDD的键和值。keys返回一个仅包含键的RDD，values返回了一个仅包含值的RDD。

val rdd= sc.parallelize(List("this is a test","hellow world ","come on "))
val words = rdd.map(x=>(x.split(" ")(0),x));
val key = words.keys
val value = words.values
key.collect().foreach(println(_))
value.collect().foreach(println(_))

his

hellow
come
this is a test
hellow world 
come on

1.转换操作reduceByKey()

reduceByKey()的功能是合并具有相同键的值，作用域是Key/Value类型的键值对，并且是只对每个Key的Value进行处理，当RDD中有许多个键相同的键值对，那么就会对这个Key的Values进行处理。

val rdd1 = sc.parallelize(List(('a',1),('d',4),('e',5),('b',2),('a',1),('b',2),('c',3)))
val r_rdd=rdd1.reduceByKey((a,b)=>a+b)
r_rdd.collect().foreach(println(_))

(d,4)

(e,5)

(a,2)

(b,4)

(c,3)

2.转换操作groupByKey()

groupByKey()是对具有相同键的值进行分组，对于一个由类型K的键和类型V的值组成的RDD，通过groupByKey()得到的RDD类型是[K,Iterable[V]]。

val rdd1 = sc.parallelize(List(('a',1),('a',4),('b',5),('b',2),('a',1),('b',2),('c',3)))
val r_rdd=rdd1.groupByKey()
r_rdd.collect().foreach(println(_))
r_rdd.map(x=>(x._1,x._2.size)).collect().foreach(println(_))
//size()用于在指定的映射中查找键/值对的数量。

(a,CompactBuffer(1, 4, 1))

(b,CompactBuffer(5, 2, 2))

(c,CompactBuffer(3))

(a,3)

(b,3)

(c,1)

3.join()连接两个RDD

连接方式（对于学过数据库SQL的人来说比较容易理解）：

（1）join

join是根据键对两个RDD进行内连接，将两个RDD中键相同的数据的值存在一个元组中，最后只返回两个RDD都存在的键的连接结果。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5)))
val j_rdd = rdd1.join(rdd2)
j_rdd.collect().foreach(println(_))

(a,(1,1))

（2）rightOuterJoin

rightOuterJoin是根据键对两个RDD进行右外连接，连接结果返回第二个RDD的所有键的连接结果，不管在第一个RDD中是否存在。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5)))
val r_rdd = rdd1 rightOuterJoin rdd2
r_rdd.collect().foreach(println(_))

d,(None,4))

(e,(None,5))
(a,(Some(1),1))

（3）leftOuterJoin

leftOuterJoin是对两个RDD的键进行左外连接的方法，与rightOuterJoin相反。返回结果保留第一个RDD的所有键。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5)))
val l_rdd = rdd1 leftOuterJoin rdd2
l_rdd.collect().foreach(println(_))

(a,(1,Some(1)))

(b,(2,None))
(c,(3,None))

（4）fullOuterJoin

fullOuterJoin是一种全外连接，会保留两个连接的RDD中所有键的连接结果。

val rdd1 = sc.parallelize(List(('a',1),('b',2),('c',3)))
val rdd2 = sc.parallelize(List(('a',1),('d',4),('e',5)))
val f_rdd = rdd1 fullOuterJoin rdd2
f_rdd.collect().foreach(println(_))

(d,(None,Some(4)))

(e,(None,Some(5)))
(a,(Some(1),Some(1)))
(b,(Some(2),None))
(c,(Some(3),None))

4.zip组合两个RDD

zip函数用于将两个RDD组合成Key/Value形式的RDD，这里要求两个RDD的partition数量以及元素数量都相同，否则会抛出异常。

val rdd1 = sc.parallelize(List(1,2,3,4,5))
val rdd2 = sc.parallelize(List('a','c','e','d','w'))
rdd1.zip(rdd2).collect().foreach(println(_))
rdd2.zip(rdd1).collect().foreach(println(_))

(1,a)

(2,c)

(3,e)

(4,d)

(5,w)

(a,1)

(c,2)

(e,3)

(d,4)

(w,5)

5.combineByKey合并相同键的值

combineByKey是Spark中一个比较核心的高级函数，其他一些高阶键值对函数底层都是用它来实现的。

combineByKey用于将相同键的数据聚合，并且允许返回类型与输入数据类型不同的返回值，combineByKey函数的定义为：

def combineByKey[C](

     createCombiner: V => C,

     mergeValue: (C, V) => C,

     mergeCombiners: (C, C) => C): RDD[(K, C)] = self.withScope {

               /*content*/

 }

以上三个重要的参数：

（1）createCombiner：V=>C，V是键值对RDD中的值部分，将该值转换为另一种类型C，C会作为每一个键的累加器的初始值。

（2）mergeValue: (C, V) => C,该函数把元素V合并到之前的元素C（createCombiner）上（这个操作在每个分区进行）。

（3）mergeCombiners：(C, C)=>C，该函数把两个元素C合并（这个操作在不同分区间进行）。

由于聚合操作会遍历分区中所有的元素，因此每个元素的键只有两种情况：以前没出现过或以前出现过。

（1）如果以前没出现过，则执行的是createCombiner方法，createCombiner()会在新遇到的键对应的累加器中赋予初始值，否则执行mergeValue方法。

（2）对于已经出现过的键，调用mergeValue来进行聚合操作，对该键的累加器对应的当前值（C个数）与新值（V格式）进行合并。

（3）由于每个分区都是独立处理的，因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器，就需要使用用户提供的mergeCombiners（）方法将各个分区的结果进行合并。

本文主要参考Spark大数据技术与运用一书。