Spark基础 --RDD详解

  1. 云栖社区>
  2. 博客>
  3. 正文

Spark基础 --RDD详解

五维空间s 2018-01-17 18:18:31 浏览847
展开阅读全文

RDD算子分为两类:Transformation和Action,如下图,记住这张图,走遍天下都不怕。


6162d728968a06809e54ecff4c9ef92208371dd6

Transformation:将一个RDD通过一种规则映射为另外一个RDD。

Action:返回结果或保存结果。

注意:只有action才触发程序的执行,transformation不触发执行。


RDD的操作种类有多个,分为: 单指RDD操作、Key/Value RDD操作、多个RDD联合操作,其他操作。


单值RDD


1. Map

 map (f: T => U) : RDD[U] ,其中f定义了类型为T的元素到类型为U

的元素的映射,RDD[T] => RDD[U]的变换

a975d77780bc53c68d691511405d2958562729e3

举例:

            var rdd=sc.makeRDD(1 to 7,3)

简写为  rdd.map(_+1)   //rdd.map(x=>x+1)


c638b233ef6cf24a47e99ac5826c9b5ec1619566


2. collect

collect(): Array[T],T是RDD中元素类型,将RDD转化为数组。

eac7b65d68ac3365ac2c9337af2a9553b22f39a0

举例:

val rdd = sc.makeRDD(1 to 7, 3)

rdd.collect()

注意:此算子非常危险,他会将所有RDD中的数据汇总到Drive端的JVM内存中,对Drive端压力很大。


3. take

take(num: Int): Array[T] ,其中k是整数,T是RDD中元素类型,返回RDD中前k个元素,并保存成数组

1d1149f3d4df566bf984d57b39345af948a24ea9

举例:

val rdd = sc.makeRDD(1 to 7, 3)

rdd.take(2)

dc68720699bc06af29e0013659bd3f8d31cb6d50


4. glom

glom() : RDD[Array[T]],将RDD中每个partition中元素转换为数组

2711db919a580061cc9a3d59e52faa79f6b73dc3

举例:

val rdd = sc.makeRDD(1 to 7, 3)

rdd.glom.collect

849ad259e2219d47ea2e120a086d7982ab1b8a52


5. coalesce

coalesce(numPartitions: Int) : RDD[T],将RDD中的partition个数合并为numPartitions个

4a3ce467897342145cef46ca96ef1c96827f1f93

举例:

val rdd = sc.makeRDD(1 to 7,7)

rdd.coalesce(3) // 生成新的RDD,它包含三个Partition


6. repartition

repartition(numPartitions: Int) :RDD[T],将RDD中的partition个数均匀合并为numPartitions个

f3ade082fb989c0802ccd36c8c1c453b121c9c4f

举例:

val list = Seq(Seq(),Seq(),Seq(),Seq(),Seq(),Seq(),

Seq(1,2,3,4,5,6,7))

val rdd = sc.makeRDD(list, 7).flatMap(x => x)

rdd.repartition(3) // 生成新的RDD,它包含三个Partition


7. filter

filter(f: T => Boolean):

RDD[T] ,其中f定义了类型为T的元素是否留下,过滤输入RDD中的元素,将f返回true的元素留下

c3b9f54390901438e667a938699f03a410da2239

举例:

var rdd=sc.makeRDD(1 to 7,7)

rdd.filter(_%3==0)

bb724ec00ec075f8449649979fef74c31fa807bf


 8. count

count(): Long,统计RDD中元素个数,并返回Long类型

2cf64d32ecc75d8b919c7d4447a6750ab8e8fcaa

val rdd = sc.makeRDD(1 to 7, 3)

rdd.count() // 统计RDD中元素总数


9. flatMap

flatMap(f: T =>TraversableOnce[U]): RDD[U],将函数f作用在RDD中每个元素上,并展开(flatten)

输出的每个结果, flatMap = flatten + map,先映射(map),再拍扁(flatten

e87b5eed666a0bca232f9e8f88a32d9eb1a4b5f0

举例:

val rdd = sc.makeRDD(1 to 3, 3)

rdd.flatMap( x => 1 to x) // 将x映射成1~x


10. reduce

reduce(f: (T, T) => T): T, 按照函数f对RDD中元素,进行规约

a46509b347756e3fd5172f2483da1237bb72bf1b

举例:

val rdd = sc.makeRDD(1 to 7, 3)

rdd.reduce((x, y) => x + y)

简写为:rdd.reduce(_ + _)

bcd14168134faa0f1edd715738bd36bd1e9f12d2


11. foreach

foreach(f: T => Unit):Unit,对RDD中每个元素,调用函数f

f80f059417c4d12d0dc20e7427ea056d05c7dd1e

举例:

val rdd = sc.makeRDD(1 to 7, 3)

rdd.foreach( x => println(x))

简写为:rdd.foreach(println)


Key/Value RDD


首先先来看下如何创建一个Key/Value的rdd

var seq=Seq((A,1),(B,1),(C,1))

var rdd=sc.makeRDD(seq)


1. mapValues

对vaule做map操作

dc834cd4d7468f74a3b9f7d698e02e7be3aaaad1

举例:

val pairs = Seq((A,1), (B,2), (A,2), (C, 4), (B, 1), (B, 1), (D, 1))

val rdd = sc.makeRDD(pairs, 3)

rdd.mapValues(_ + 1)

59da49ff2804b636a101df40f4db0664b846abbc


2. reduceByKey

对Key相同的value做计算

c7367af8a9867daf571d5a6f9b3abbd8b037a48c

举例:

val pairs = Seq(('A',1), ('B',2), ('A',2), ('C', 4), ('B', 1), ('B', 1), ('D', 1))

val rdd = sc.makeRDD(pairs, 3)

rdd.reduceByKey(_ + _)

7840fab36f17934175fb30e994ae60806ac98f61


3. groupByKey

将RDD[key,value] 按照相同的key进行分组,形成RDD[key,Iterable[value]]的形式, 有点类似于sql中的groupby

52f76378b2f9aa472bbbdfd8aeba181bf4ae5c3a

举例:

val pairs = Seq((A,1), (B,2), (A,2), (C, 4), (B, 1), (B, 1), (D, 1))

val rdd = sc.makeRDD(pairs, 3)

rdd.groupByKey()


注意:能用reducebykey代替就不用groupbykey,groupbykey会将所有的元素进行聚合,消耗大量内存。


多RDD


1. union

将多个RDD合并为一个RDD

a571aa725cb66d976c234dbdb62d16c0a4d2612b

举例:


val pairs1 = Seq((A,1), (B,1), (C,1), (D, 1), (A, 2), (C, 3))

val rdd1 = sc.makeRDD(pairs1, 3)

val pairs2 = Seq((A,4), (D,1), (E, 1))

val rdd2 = sc.makeRDD(pairs2, 2)

rdd1.union(rdd2)


2. zip

zip函数用于将两个RDD组合成Key/Value形式的RDD,如果两个rdd中的partition数量不一致,会报错。

a404057613ec8fc28b8e7a5406c5b16a85c69dd1

举例:

val s1 = Seq(A, B, C, D, E)

val rdd1 = sc.makeRDD(s1)

val s2 = Seq(1, 2, 3, 4, 5)

val rdd2 = sc.makeRDD(s2)

rdd1.zip(rdd2)

27142c9b418b24d74439595e40a082c448be53fd


3. join

join相当于SQL中的内关联join,只返回两个RDD根据K可以关联上的结果,join只能用于两个RDD之间的关联,

如果要多个RDD关联,多关联几次即可

fd5f7efb80d13a7a932022c10b1c512bb272da3a

举例:

val pairs1 = Seq((A,1), (B,1), (C,1), (D, 1), (A, 2), (C, 3))

val rdd1 = sc.makeRDD(pairs1, 3)

val pairs2 = Seq((A,4), (D,1), (C,1), (E, 1))

val rdd2 = sc.makeRDD(pairs2, 2)

rdd1.join(rdd2)


还有些是是其他rdd操作符,这里就不讲解了,上述所写如有不对之处,还请各位前辈赐教。



网友评论

登录后评论
0/500
评论
五维空间s
+ 关注