第17课：RDD案例(常用算子解释)

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1、创建RDD，代码如下：

1. def sparkContext(name:String) = {
   
2.     val conf = new SparkConf().setAppName(name).setMaster("local")
   
3.     val sc = new SparkContext(conf)
   
4.     sc
   
5. }

复制代码

2、Map：适用于任何集合，且对其作用的集合中的每一个元素循环遍历，并调用其作为参数的函数对每一个遍历的元素进行具体化处理。代码如下：

1. def mapTransformation(sc:SparkContext): Unit ={
   
2.     val nums = sc.parallelize(1 to 10)    //根据集合创建RDD
   
3.     val mapped = nums.map(item=> 2 * item)
   
4.     mapped.collect.foreach(println)
   
5. }

复制代码

结果如下所示：

3、Filter：遍历集合中的所有元素，将每个元素作为参数放入函数中进行判断，将判断结果为真的元素筛选出来。代码如下：

1. def filterTransformation(sc:SparkContext): Unit ={
   
2.     val nums = sc.parallelize(1 to 20)    //根据集合创建RDD
   
3.     val filtered = nums.filter(item => item % 2 == 0)
   
4.     filtered.collect.foreach(println)
   
5. }

复制代码

结果如下所示：

4、Flatmap：通过传入的作为参数的函数来作用与RDD的每个字符串进行单词切分，然后把切分后的结果合并成一个大的集合。代码如下：

1. def flatmapTransformation(sc:SparkContext): Unit ={
   
2.     val bigData = Array("scala","spark","java Hadoop","java tachyon")
   
3.     val bigDataString =sc.parallelize(bigData)
   
4.     val words= bigDataString.flatMap(line=>line.split(" "))
   
5.    words.collect.foreach(println)
   
6. }

复制代码

结果如下所示：

5、groupByKey：将传入的tuple数组生成为RDD，通过groupByKey方法将RDD通过key进行分组汇总，并生成一个新的RDD。代码如下：

1. def groupByKeyTransformation(sc:SparkContext): Unit ={
   
2.     val data =                 Array(Tuple2(100,"Spark"),Tuple2(100,"Tachyon"),Tuple2(90,"Hadoop"),Tuple2(80,"Kafka"),Tuple2(70,"Scala"))
   
3.     val dataRDD = sc.parallelize(data)
   
4.     val group = dataRDD.groupByKey()
   
5.     group.collect.foreach(pair=>println(pair._1+":"+pair._2))
   
6. }

复制代码

结果如下所示：

6、reduceByKey：对key相同的元素进行value值得相加。代码如下：

1. def reduceByKeyTransformation(sc:SparkContext): Unit ={
   
2.     val lines =sc.textFile("C://Users//feng//IdeaProjects//WordCount//src//SparkText.txt",1)
   
3.     val reduce= lines.map(line=>(line,1)).reduceByKey(_+_)
   
4.     reduce.collect.foreach(pair=>println(pair._1+":"+pair._2))
   
5. }

复制代码

文件内容如下所示：

结果如下所示：

7、join：根据相同key，把不同的RDD合并为一个RDD。代码如下：

1. def joinTransformation(sc:SparkContext): Unit ={
   
2.     //大数据中最重要的算子
   
3.     val studentNames=Array(
   
4.         Tuple2(1,"Spark"),
   
5.         Tuple2(2,"Tachyon"),
   
6.         Tuple2(3,"Hadoop")
   
7.   )
   
8.     val studentScore=Array(
   
9.         Tuple2(1,100),
   
10.         Tuple2(2,95),
    
11.         Tuple2(3,65),
    
12.         Tuple2(2,95),
    
13.         Tuple2(3,65)
    
14.   )
    
15.     val names = sc.parallelize(studentNames)
    
16.     val scores = sc.parallelize(studentScore)
    
17.     val studentNameAndScore=names.join(scores)
    
18.     studentNameAndScore.collect.foreach(println)
    
19. }

复制代码

结果如下所示：

8、cogroup：协同分组，首先将两个RDD的内容进行join,在此基础上，以ID为key的情况下将改ID内容的所有分数聚合到一起。代码如下：

1. def cogroupTransformation(sc:SparkContext): Unit ={
   
2.     val nameList = Array(
   
3.         Tuple2(1,"Spark"),
   
4.         Tuple2(2,"Scala"),
   
5.         Tuple2(3,"Hadoop")
   
6.   )
   
7.     val scoreList = Array(
   
8.         Tuple2(1,100),
   
9.         Tuple2(2,90),
   
10.         Tuple2(3,87),
    
11.         Tuple2(1,80),
    
12.         Tuple2(2,90),
    
13.         Tuple2(2,60)
    
14.   )
    
15.     val names = sc.parallelize(nameList)
    
16.     val scores =sc.parallelize(scoreList)
    
17.     val nameScores= names.cogroup(scores)
    
18.     nameScores.collect.foreach(println)
    
19. }

复制代码

结果如下所示：

注：本学习笔记来自DT大数据梦工厂

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏1 回帖

关键词：transform formation Parallel Students filtered Spark scala DT_Spark 大数据