其他分享
首页 > 其他分享> > Spark追妻系列(Value类型的RDD)

Spark追妻系列(Value类型的RDD)

作者:互联网

今天是大年初三,猴赛雷

小谈

        这几天每天晚上给她发一个红包,拜年红包而且还可以添加表情包。感觉现在过年好没有年味吖。嗑瓜子磕的嗓子都疼了。

        Spark中的算子有很多,有Value类型,双Value类型,这两天写的都是Value类型的,昨天讲的是关于map的映射。

        今天讲剩余的算子

glom

        glom算子将RDD中的每一个分区变成一个数组,并放置在RDD中,数组中的元素类型与原分区中的类型相同。原本这个分区里面的数是分散的,glom之后,这个分区里面的元素会变成一个数组。

def glom(): RDD[Array[T]] = withScope { 
new MapPartitionsRDD[Array[T], T]
(this, (_, _, iter) => Iterator(iter.toArray))

        可以看出来,glom底层用到了MapPartitionsRDD对象,这个MapPartitionsRDD对象底层重写了getPartitions方法,这个方法用到了RDD的依赖,保持分区的数据不变,然后将分区的数据转变成数组类型。

        来一个例子,计算所有分区最大值的求和。

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD") 
val sparkContext = new SparkContext(sparkConf) 
// 两个分区 12 在分区1 35 在分区2 
val value = sparkContext.makeRDD(List(1, 2, 3, 5), 2) 
//对每个分区的数组glom 
val value1 = value.glom() 
//找到每个数组的最大值 
val value2 = value1.map(array => array.max) 
//将每个数组最大值进行求和 
val sum = value2.collect().sum

        分区最大值求和

        分区1的最大值: 2 分区2的最大值: 5

        collect之后,每个节点给Driver返回每个数组的最大值,之后相加

groupBy

        将数据按照指定的规则进行分组,分区默认不变,但是可能分组后的数据不再原来的分区,数据会被打乱,这样的操作就是Shuffle。极限情况下,数据可能分在同一个分区。

        将相同的key的数据分到一个迭代器里面。

        下面举一个例子

 

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
//将数据按照指定的规则进行分组 
val value = sparkContext.makeRDD(List(1, 2, 3, 4), 2) 
val value1 = value.groupBy(_ % 2 == 0) 
value1.collect().foreach(println(_))

        刚开始 两个分区

        分区1: 1 2 分区2:3 4

        group by之后,数据发生改变,只有一个分区存在数据

(1,(false,CompactBuffer(1, 3))) (1,(true,CompactBuffer(2, 4)))

        可以看到,分区后的数据并不是上面图解的那样,因为发生了Shuffle

        为了体现出group by的作用,下面将将 List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
val value = sparkContext.makeRDD(List("Hello", "hbase", "Hive", "Hadoop"), 2) 
val value1 = value.groupBy(_.charAt(0)) 
value1.collect().foreach(println(_)) }

        看一下结果

(h,CompactBuffer(hbase)) (H,CompactBuffer(Hello, Hive, Hadoop))

        前面的字符是我们分组的依据,后面的迭代器就是分组后的元素

WordCount

        学习mr的时候,第一个例子就是求WordCount。下面也用Spark来进行操作这就是简易的wc。首先将每个单词分开,分开之后对每个单词进行计数(单词,1),计完数后,根据单词分组,单词分组之后求迭代器的大小。

​
val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
val value = sparkContext.makeRDD(List("Hello Spark", "Hello Scala"), 2) 
//先把单词分成一个一个 
val value1 = value.flatMap(_.split(" ")) 
//对每个单词进行计数 
val value2 = value1.map((_, 1)) 
//分组 
val value3 = value2.groupBy(_._1) 
val value5 = value3.map {
 case (word, iter) => (word, iter.size) } 

value5.collect().foreach(println(_))

​

Filter

        将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合的丢弃。对数据进行筛选之后,分区不变,但是分区内的数据可能不均衡,可能会造成数据倾斜。

 

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
val value = sparkContext.makeRDD(List(1, 2, 3, 4)) 
val value1 = value.filter(_ != 1) 
value1.collect().foreach(println(_))

Distinct

        对数据集中的重复数据进行去重

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
val value = sparkContext.makeRDD(List(1, 2, 3, 4,4)) 
val value1 = value.distinct()

        如果在distinct中没有参数,那么最后的结果就是1,2,3,4

        如果在distinct中有参数,比如2,那么最后的结果就会有两个分区,原本一个分区的(因为设置的setMaster为Local,所以默认一个分区),原本一个分区的数据,分散到两个分区里面,发生了shuffle

coalesce

        根据数据量缩减分区。如果分区数过多,可以进行减少分区。

        coalesce这个算子里面的参数有一个 shuffle ,如果为True,那么这个coalesce就会将数据shuffle,会将数据进行打乱,如果为False,那么数据就不会被打乱。

        先来看看例子把

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
//四个数据,三个分区 1 2 34 
val value = sparkContext.makeRDD(List(1, 2, 3, 4),3) 
//减少分区为两个,shuffle默认为False,不会打乱数据 
val value1 = value.coalesce(2) 
//可以来看数据的分区 
val value2 = value1.mapPartitionsWithIndex(
(index, iter) => { iter.map(num => (index, num)) })
value2.collect().foreach(println(_))

        首先看一下先前的分区

        分区1:1 分区2 :2 分区3:3 4

        现在看一下在shuffle参数为False的情况下,分区的情况

        可以看到,分区真的减少了,明明数据已经不再第三个分区了,跑到第二个分区了,这数据不就被打乱了。

        其实并不是,看一看 分区3的数据 3和4 一起进入了第二个分区,第三分区的数据没有被打乱。这就是所谓的shuffle,并没有将数据打乱,只是将原来的数据放到了其它的分区。

        现在看一下shuffle参数为True的情况下

        如果有Shuffle了,1 2 3 4 分到两个分区里面,

        分区1 : 1 3 分区2: 2 4

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
//四个数据,三个分区 1 2 34 
val value = sparkContext.makeRDD(List(1, 2, 3, 4),3) 
//减少分区为两个,shuffle参数为true,会有shuffle阶段 
val value1 = value.coalesce(2,true) 
//可以来看数据的分区 
val value2 = value1.mapPartitionsWithIndex(
(index, iter) => { iter.map(num => (index, num)) })
value2.collect().foreach(println(_))

        可以看到,原来的第一分区,第二分区里面的数据不改变,只是将第三个分区里面的数据放入到两个分区里面

repartition

        repartition算子和coalesce算子基本相同,不过reaprtition算子可以将分区数目减少,也可以将分区数目增大,不论是增大或减小,repartition算子一定会有shuffle阶段,一定会打乱数据。

 

        可以看到,shuffle参数默认就是true。

        就是会改变数据的分区

sortBy

        该操作用于排序数据。在排序之前,可以将数据通过 f 函数进行处理,之后按照 f 函数处理

的结果进行排序,默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一

致。中间存在 shuffle 的过程

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
val value = sparkContext.makeRDD(List(4, 3, 1, 2)) 
val value1 = value.sortBy(num => num) 
value1.collect().foreach(println(_))

最后结果

 

可以看到是升序排序的,如果想降序排序,就添加一个参数false

        默认是true,按照升序排序

总结

        今天总算将Value类型的算子讲完了,明天会将双Value类型的算子讲解完毕。

        今年是大年初三,过年也要学习哦

标签:sparkContext,value,val,分区,追妻,RDD,Value,value1,new
来源: https://blog.csdn.net/weixin_46300771/article/details/122777110