良许Linux教程网 干货合集 SparkValue类型的常用算子

SparkValue类型的常用算子

Spark RDD常用算子:Value类型

Spark之所以比Hadoop灵活和强大,其中一个原因是Spark内置了许多有用的算子,也就是方法。通过对这些方法的组合,编程人员就可以写出自己想要的功能。说白了spark编程就是对spark算子的使用,下面为大家详细讲解一下SparkValue类型的常用算子

SparkValue类型的常用算子

map

函数说明:

map() 接收一个函数,该函数将RDD中的元素逐条进行映射转换,可以是类型的转换,也可以是值的转换,将函数的返回结果作为结果RDD编程。

函数签名:

def map[U: ClassTag](f: T => U): RDD[U]

案例演示

   val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
   val sc = new SparkContext(sparkConf)
   //算子 -map
   val rdd = sc.makeRDD(List(1, 2, 3, 4),2)
   val mapRdd1 = rdd.map(
     _*2
   )
   mapRdd1.collect().foreach(println)
   sc.stop()

运行结果

2
4
6
8

mapPartitons

函数说明:

将待处理的数据以分区为单位发送到待计算节点上进行处理,mapPartition是对RDD的每一个分区的迭代器进行操作,返回的是迭代器。这里的处理可以进行任意的处理。

函数签名:

def mapPartitions[U: ClassTag](
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

案例演示

 def main(args: Array[String]): Unit = {
   val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
   val sc = new SparkContext(sparkConf)
   //算子 -mapPartitons 计算每个分区的最大数
   val rdd = sc.makeRDD(List(1, 34, 36,345,2435,2342,62,35, 4),4)
   val mapParRdd = rdd.mapPartitions(
     iter => {
       List(iter.max).iterator
     }
   )
   mapParRdd.foreach(println)
   sc.stop()
 }

运行结果:

62
2435
34
345

mapPartitonsWithIndex

函数说明:

将待处理的数据以分区为单位发送到计算节点上,这里的处理可以进行任意的处理,哪怕是过滤数据,在处理的同时可以获取当前分区的索引值。

函数签名:

def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

案例演示:

  1. 将数据进行扁平化映射并且打印所在的分区数
def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List("Hello Spark""Hello Scala""Word Count"),2)
   val mapRDD = rdd.flatMap(_.split(" "))
   val mpwiRdd = mapRDD.mapPartitionsWithIndex(
     (index, datas) => {
       datas.map(
         num => {
           (index, num)
         }
       )
     }
   )
   mpwiRdd.collect().foreach(println)
 }

运行结果:

(0,Hello)
(0,Spark)
(1,Hello)
(1,Scala)
(1,Word)
(1,Count)
  1. 将数据进行扁平化映射只打印所在第一分区的数据
def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List("Hello Spark""Hello Scala""Word Count"),2)
   val mapRDD = rdd.flatMap(_.split(" "))
   val mpwiRdd = mapRDD.mapPartitionsWithIndex(
     (index, datas) => {
       if (index==0){
         datas.map(
           num => {
             (index, num)
           }
         )
       }else{
       Nil.iterator
       }
     }
   )
   mpwiRdd.collect().foreach(println)

运行结果:

(0,Hello)
(0,Spark)

flatMap

函数说明:

将数据进行扁平化之后在做映射处理,所以算子也称为扁平化映射

函数签名:

def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

案例演示:

将每个单词进行扁平化映射

def main(args: Array[String]): Unit = {
 val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
 val sc = new SparkContext(sparkConf)
 //算子 -map
 val rdd = sc.makeRDD(List("Hello Scala","Hello Spark"), 2)
 val FltRdd = rdd.flatMap(
   _.split(" ")
 )
 FltRdd.foreach(println)
 sc.stop()
}

运行结果:

Hello
Scala
Hello
Spark

glom

函数说明:

glom的作用就是将一个分区的数据合并到一个array中。

函数签名:

def glom(): RDD[Array[T]]

案例演示:

  1. 将不同分区rdd的元素合并到一个分区
 def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9),2)
   val glomRdd = rdd.glom()
   glomRdd.collect().foreach(data=>println(data.mkString(",")))
   sc.stop()
 }

运行结果:

1,2,3,4
5,6,7,8,9

groupBy

函数说明:

将数据根据指定的规则进行分组,分区默认不变,单数数据会被打乱,我们成这样的操作为shuffer,

函数签名:

def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

案例演示:

  1. 按照奇偶数进行groupby分区
 def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8,10),2)
   val groupByRDD = rdd.groupBy(_ % 2 == 0)
   groupByRDD.collect().foreach(println)
   sc.stop()
 }

运行结果:

(false,CompactBuffer(1, 3, 5, 7))
(true,CompactBuffer(2, 4, 6, 8, 10))
  1. 按照单词的首字母进行分组
 def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List("Hello","Tom","Timi","Scala","Spark"))
   val groupByRDD = rdd.groupBy(_.charAt(0))
   groupByRDD.collect().foreach(println)
   sc.stop()
 }

运行结果:

(T,CompactBuffer(Tom, Timi))
(H,CompactBuffer(Hello))
(S,CompactBuffer(Scala, Spark))

filter

函数说明:

filter即过滤器的意思,所以filter算子的作用就是过滤的作用。filter将根据指定的规则进行筛选过滤,符合条件的数据保留,不符合的数据丢弃,当数据进行筛选过滤之后,分区不变,但分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜。

函数签名:

def filter(f: T => Boolean): RDD[T]

案例演示:

  1. 筛选出能被二整除的数字
 def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List(46,235,246,2346,3276,235,234,6234,6245,246,24,6246,235,26,265))
   val filterRDD = rdd.filter(_ % 2 == 0)
   filterRDD.collect().foreach(println)
   sc.stop()
 }

运行结果:

46
246
2346
3276
234
6234
246
24
6246
26

2.筛选单词中包含H的

 def main(args: Array[String]): Unit = {
   val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
   val sc = new SparkContext(conf)
   val rdd = sc.makeRDD(List("Hello","Horber","Hbeer","ersfgH","Scala","Hadoop","Zookeeper"))
   val filterRDD = rdd.filter(_.contains("H"))
   filterRDD.collect().foreach(println)
   sc.stop()
 }

运行结果:

Hello
Horber
Hbeer
ersfgH
Hadoop

以上就是良许教程网为各位朋友分享的Linu系统相关内容。想要了解更多Linux相关知识记得关注公众号“良许Linux”,或扫描下方二维码进行关注,更多干货等着你 !

img
本文由 良许Linux教程网 发布,可自由转载、引用,但需署名作者且注明文章出处。如转载至微信公众号,请在文末添加作者公众号二维码。
良许

作者: 良许

良许,世界500强企业Linux开发工程师,公众号【良许Linux】的作者,全网拥有超30W粉丝。个人标签:创业者,CSDN学院讲师,副业达人,流量玩家,摄影爱好者。
上一篇
下一篇

发表评论

邮箱地址不会被公开。 必填项已用*标注

联系我们

联系我们

公众号:良许Linux

在线咨询: QQ交谈

邮箱: yychuyu@163.com

关注微信
微信扫一扫关注我们

微信扫一扫关注我们

关注微博
返回顶部