文档章节

Spark(四):RDD算子分类。

berg-dm
 berg-dm
发布于 2016/06/03 18:50
字数 2455
阅读 2199
收藏 7

RDD算子分类,大致可以分为两类,即:

1.  Transformation:转换算子,这类转换并不触发提交作业,完成作业中间过程处理。

2.  Action:行动算子,这类算子会触发SparkContext提交Job作业。

 

一:Transformation:转换算子

1.  map:

    将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素。源码中map算子相当于初始化一个RDD,新RDD叫做MappedRDD(this,sc.clean(f) )。即:

    map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。 任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at :27

scala> val b = a.map(x => x*3)
b: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[7] at map at :29

scala> a.collect
res7: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)                             

scala> b.collect
res8: Array[Int] = Array(3, 6, 9, 12, 15, 18, 21, 24, 27)

  上述例子中把原RDD中每个元素都乘以3来产生一个新的RDD。

2.  mapPartitions:

    mapPartitions函数获取到每个分区的迭代器,在函数中通过这个分区整体的迭代器对整个分区的元素进行操作。内部实现是生成MapPartitionsRDD。

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[10] at parallelize at :27

scala> a.collect
res11: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)                            

scala> var c = a.mapPartitions( a=>a.filter(_>=7) )
c: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[11] at mapPartitions at :29

scala> c.collect
res12: Array[Int] = Array(7, 8, 9) 

上述例子是通过函数filter对分区中所有数据进行过滤。

3.  mapValues

        针对(key,value)型数据中的Value进行操作,而不对Key进行处理。即:

        mapValues顾名思义就是输入函数应用于RDD中Kev-Value的Value,原RDD中的Key保持不变,与新的Value一起组成新的RDD中的元素。因此,该函数只适用于元素为KV对的RDD。

scala> val a = sc.parallelize(List("Hadoop","HBase","Hive","Spark"), 2)
a: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[12] at parallelize at :27

scala> val b = a.map(x => (x.length,x) )
b: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[13] at map at :29

scala> b.mapValues(""+_+"").collect
res14: Array[(Int, String)] = Array((6,Hadoop), (5,HBase), (4,Hive), (5,Spark))

4.  mapWith:

    mapWith是map的另外一个变种,map只需要一个输入函数,而mapWith有两个输入函数。

eg: 把partition index 乘以10,然后加上2作为新的RDD的元素.(3 是将十个数分为三个区)

scala> val x  = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10),3)

scala> x.mapWith( a => a*10 )( (a,b)=>(b+2)).collect
res16: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)

5.  flatMap:

    将原来RDD中的每个元素通过函数f转换为新的元素,并将生成的RDD的每个集合中的元素合并为一个集合,内部创建FlatMappedRDD(this,sc.clean() )。即:

    与map类似,区别是原RDD中的元素经map处理后只能生成一个元素,而原RDD中的元素经flatmap处理后可生成多个元素来构建新RDD。

eg:对原RDD中的每个元素x产生y个元素(从1到y,y为元素x的值)。

scala> val a = sc.parallelize(1 to 4,2)

scala> val b = a.flatMap(x => 1 to x )

scala> a.collect
res17: Array[Int] = Array(1, 2, 3, 4)                                           

scala> b.collect
res18: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4)

6.  flatMapWith:

        flatMapWith与mapWith很类似,都是接收两个函数,一个函数把partitionIndex作为输入,输出是一个新类型A;另外一个函数是以二元组(T,A)作为输入,输出为一个序列,这些序列里面的元素组成了新的RDD。

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9),3)

scala> a.collect
res0: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)                             

scala> a.flatMapWith(x => x,true)((x,y)=>List(y,x)).collect
res1: Array[Int] = Array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2, 8, 2, 9)

7.  flatMapWithValues:

        flatMapValues类似于mapValues,不同的在于flatMapValues应用于元素为KV对的RDD中Value。每个一元素的Value被输入函数映射为一系列的值,然后这些值再与原RDD中的Key组成一系列新的KV对。

scala> val a = sc.parallelize( List((1,2),(3,4),(3,6)) )

scala> a.collect
res2: Array[(Int, Int)] = Array((1,2), (3,4), (3,6))

scala> val b = a.flatMapValues( x => x.to(5))

scala> b.collect
res3: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))

上述例子中原RDD中每个元素的值被转换为一个序列(从其当前值到5),比如第一个KV对(1,2), 其值2被转换为2,3,4,5。然后其再与原KV对中Key组成一系列新的KV对(1,2),(1,3),(1,4),(1,5)。

8.  reduce:

        reduce将RDD中元素两两传递给输入函数,同时产生一个新的值,新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止。

eg:对元素求和。

scala> val a = sc.parallelize(1 to 10 )

scala> a.reduce( (x,y) => x + y )
res5: Int = 55

9.  reduceByKey

        顾名思义,reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行reduce,因此,Key相同的多个元素的值被reduce为一个值,然后与原RDD中的Key组成一个新的KV对。

eg:对Key相同的元素的值求和,因此Key为3的两个元素被转为了(3,10)。

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))

scala> a.reduceByKey((x,y)=>x+y).collect
res6: Array[(Int, Int)] = Array((1,2), (3,10))

10.  cartesian:

        对两个RDD内的所有元素进行笛卡尔积操作(耗内存),内部实现返回CartesianRDD。

scala> val a = sc.parallelize(List(1,2,3))

scala> val b = sc.parallelize(List(4,5,6))

scala> val c = a.cartesian(b)

scala> c.collect
res15: Array[(Int, Int)] = Array((1,4), (1,5), (1,6), (2,4), (3,4), (2,5), (2,6), (3,5), (3,6))

11.  Sample:

        sample将RDD这个集合内的元素进行采样,获取所有元素的子集。用户可以设定是否有有放回的抽样,百分比,随机种子,进而决定采样方式。

        内部实现: SampledRDD(withReplacement,fraction,seed)。
        函数参数设置:
‰             withReplacement=true,表示有放回的抽样。
‰             withReplacement=false,表示无放回的抽样。

   根据fraction指定的比例,对数据进行采样,可以选择是否用随机数进行替换,seed用于指定随机数生成器种子。

scala> val a = sc.parallelize(1 to 100,3)

scala> a.sample(false,0.1,0).count
res16: Long = 12

scala> a.sample(false,0.1,0).collect
res17: Array[Int] = Array(10, 47, 55, 73, 76, 84, 87, 88, 91, 92, 95, 98)

scala> a.sample(true,0.7,scala.util.Random.nextInt(10000)).count
res19: Long = 75

scala> a.sample(true,0.7,scala.util.Random.nextInt(10000)).collect
res20: Array[Int] = Array(1, 3, 3, 3, 5, 6, 9, 9, 9, 9, 10, 10, 15, 17, 20, 23, 23, 27, 28, 31, 32, 32, 34, 35, 36, 36, 36, 36, 38, 39, 41, 42, 42, 43, 45, 47, 49, 49, 50, 50, 51, 51, 54, 55, 55, 57, 57, 57, 57, 57, 59, 59, 61, 61, 63, 67, 72, 74, 76, 76, 80, 80, 81, 81, 81, 82, 83, 85, 87, 88, 90, 93, 95, 96, 97, 97, 99, 100)

12.  union:

        使用union函数时需要保证两个RDD元素的数据类型相同,返回的RDD数据类型和被合并的RDD元素数据类型相同。并不进行去重操作,保存所有的元素,如果想去重,可以使用distinct()。同时,spark还提供更为简洁的使用union的API,即通过++符号相当于union函数操作。

eg: a 与 b 的联合

scala> val a = sc.parallelize(List(("A",1),("B",2),("c",3),("A",4),("C",5) ))

scala> val b = sc.parallelize(List(("A",5),("B",6),("A",4),("C",9) ))

scala> a.union(b).collect
res22: Array[(String, Int)] = Array((A,1), (B,2), (c,3), (A,4), (C,5), (A,5), (B,6), (A,4), (C,9))

去重复:

scala> val d = sc.parallelize(List(("A",5),("B",6),("A",5) ))

scala> d.distinct.collect
res25: Array[(String, Int)] = Array((B,6), (A,5))

 13.  groupBy:

            将元素通过函数生成相应的Key,数据就转化为Key-Value格式,之后将Key相同的元素分为一组。

    eg:根据数据集中的每个元素的K值对数据分组

scala> val a = sc.parallelize(List(("A",1),("B",2),("c",3),("A",4),("C",5) ))

scala> a.groupByKey().collect
res21: Array[(String, Iterable[Int])] = Array((B,CompactBuffer(2)), (A,CompactBuffer(1, 4)), (C,CompactBuffer(5)), (c,CompactBuffer(3)))

14.  join:

            join对两个需要连接的RDD进行cogroup函数操作,将相同key的数据能偶放到一个分区,在cgroup操作之后形成新RDD对每个key下的元素进行笛卡尔积的操作,返回的结果在展平,对应key下的所有元组形成一个集合。最后返回 RDD[(K, (V, W))]。


    eg:a与b两个数据连接,相当于表的关联

scala> val a = sc.parallelize(List(("A",1),("B",2),("c",3),("A",4),("C",5) ))

scala> val b = sc.parallelize(List(("A",5),("B",6),("A",4),("C",9) ))

scala> a.join(b).collect
res23: Array[(String, (Int, Int))] = Array((B,(2,6)), (A,(1,5)), (A,(1,4)), (A,(4,5)), (A,(4,4)), (C,(5,9)))

15.  cache:

            cache将RDD元素从磁盘缓存到内存。相当于 persist(MEMORY_ONLY) 函数的
功能。

16.  persist:

            persist函数对RDD进行缓存操作,数据缓存在哪里,由StorageLevel这个枚举类型进行确定。DISK 代表磁盘,MEMORY 代表内存, SER 代表数据是否进行序列化存储。

            函数定义: persist(newLevel:StorageLevel)

             StorageLevel 是枚举类型,代表存储模式。

MEMORY_AND_DISK_SER 代表数据可以存储在内存和磁盘,并且以序列化的方式存储,其他同理。

二:Action:行动算子

1.  foreach:

        foreach对RDD中的每个元素都应用f函数操作,不返回 RDD 和 Array, 而是返回Uint。

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9),3)

scala> a.foreach(println(_))
4
5
6
7
8
9
1
2
3

2.  saveAsTextFile:

        函数将数据输出,存储到 HDFS 的指定目录。

        函数的内 部实现,其内部通过调用 saveAsHadoopFile 进行实现:

            this.map(x => (NullWritable.get(), new Text(x.toString)))

                            .saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)

        将 RDD 中的每个元素映射转变为 (null, x.toString),然后再将其写入 HDFS。

3.  collect:

        collect相当于toArray,不过已经过时不推荐使用,collect将分布式的RDD返回为一个单机的scala Array数据,在这个数组上运用 scala 的函数式操作。

4.  count:

        count返回整个RDD的元素个数。

scala> val a = sc.parallelize(1 to 10 )

scala> a.collect
res9: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)                         

scala> a.count
res10: Long = 10

 

更多学习:

http://www.tuicool.com/articles/ZfeQrq7

http://my.oschina.net/lgscofield/blog/497145

http://www.tuicool.com/articles/2iQVr2

《Spark大数据分析实战》

 

© 著作权归作者所有

berg-dm
粉丝 26
博文 98
码字总数 88970
作品 0
深圳
程序员
私信 提问
Spark 【数据挖掘平台介绍】 - Spark 1.1.0

一:Spark Spark已正式申请加入Apache孵化器,从灵机一闪的实验室“电火花”成长为大数据技术平台中异军突起的新锐。本文主要讲述Spark的设计思想。Spark如其名,展现了大数据不常见的“电光...

止静
2014/09/16
3K
0
Spark--大数据的“电光石火”

Spark已正式申请加入Apache孵化器,从灵机一闪的实验室“电火花”成长为大数据技术平台中异军突起的新锐。本文主要讲述Spark的设计思想。Spark如其名,展现了大数据不常见的“电光石火”。具...

一枚Sir
2014/08/15
316
1
Spark性能优化篇二: 开发调优

Spark性能优化篇二: 开发调优 1、前言 在大数据计算领域,Spark已经成为了越来越流行、越来越受欢迎的计算平台之一。Spark的功能涵盖了大数据领域的离线批处理、SQL类处理、流式/实时计算、...

u010262291
2018/05/30
0
0
高性能Spark作业基础:你必须知道的调优原则及建议

在大数据计算领域,Spark已经成为了越来越流行、越来越受欢迎的计算平台之一。Spark的功能涵盖了大数据领域的离线批处理、SQL类处理、流式/实时计算、机器学习、图计算等各种不同类型的计算操...

李雪蕤
2018/08/22
0
0
Spark Core介绍以及架构

应用场景 Apache Spark 是加州大学伯克利分校的 AMP Labs 开发的开源分布式轻量级通用计算框架。由于 Spark 基于内存设计,使得它拥有比 Hadoop 更高的性能(极端情况下可以达到 100x),并且...

wsc449
2018/01/17
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Android面试常客之Handler全解

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。 本文链接:https://blog.csdn.net/fnhfire_7030/article/details/79518819 前言:又到了一年...

shzwork
11分钟前
1
0
position sticky 定位

本文转载于:专业的前端网站➫position sticky 定位 1、兼容性 https://caniuse.com/#search=sticky chrome、ios和firefox兼容性良好。 2、使用场景 sticky:粘性。粘性布局。 在屏幕范围内时...

前端老手
18分钟前
1
0
CentOS 7 yum 安装 PHP7.3 教程

参考:https://www.mf8.biz/centos-rhel-install-php7-3/ 1、首先安装 EPEL 源: yum install epel-release 安装 REMI 源: yum install http://rpms.remirepo.net/enterprise/remi-release......

dragon_tech
33分钟前
1
0
Linux物理网卡聚合及桥接

Linux内部实现的bridge可以把一台机器上的多张网卡桥接起来,从而把自己作为一台交换机。同时,LInux bridge还支持虚拟端口,即桥接的不一定都是物理网卡接口,还可以是虚拟接口。目前主要表...

xiangyunyan
33分钟前
1
0
一起来学Java8(一)——函数式编程

在这篇文章中,我们将了解到在Java8下如何进行函数式编程。 函数式编程 所谓的函数式编程就是把函数名字当做值进行传递,然后接收方拿到这个函数名进行调用。 首先来看下JavaScript如何进行函...

猿敲月下码
今天
6
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部