常见的SparkRDD操作

谁借莪１个温暖的怀抱￠ 2021-12-03 04:43 413阅读 0赞

### 常见的SparkRDD操作 ###

**sparkRDD的操作，从宏观上分为：Transformation和Action，但是具体的还以分为输入算子、变换算子、缓存算子，以及行动算子。**

##### 1.Transformation算子 #####

### 1.1. flatMap ###

是Spark RDD中的转换算子，对RDD中的每一个元素都执行，前后元素的对应关系是one-2-many。也就是说，对一个元素执行RDD的操作，可以产生多个元素。

/**
          * 主要作用就是做拆分
          *    one-2-many
          *   将字符串进行单词的拆分
          */
        def flatMapOps: Unit = {
            val conf = new SparkConf()
                    .setAppName("SparkTransformation")
                    .setMaster("local[*]")
            val sc = new SparkContext(conf)
            val list = List(
                "yan xin xin",
                "zhai zhao pin"
            )
    
            val listRDD = sc.parallelize(list)
    
            val wordsRDD = listRDD.flatMap(line => line.split("\\s+"))
            wordsRDD.foreach(println)
            sc.stop()
        }

### 1.2. map ###

是Spark RDD中的转换算子，对RDD中的每一个元素都执行，前后元素的对应关系是one-2-one。也就是说，对一个元素执行RDD的操作，可以产生一个元素。

/**
          * 将集合中的每一个元素*7
          */
        def mapOps(sc: SparkContext): Unit = {
            val list = 1 to 7
    
            val listRDD = sc.parallelize(list)
            val retRDD = listRDD.map(_ * 7)
            retRDD.foreach(println)
        }

### 1.3. filter ###

对RDD中的每一个元素执行对应的func函数，保留该func函数返回值为true的元素，组成一个新的RDD，过滤掉返回值为false的元素。

/*
            过滤掉班级中的男生，只留下小姐姐
            stu
                id  name    age     gender
         */
        def filterOps(sc: SparkContext): Unit = {
    
            val stuList = List(
                "1 候雪杰  18  0",
                "2 扈紫豪  28  1",
                "3 张钟方  16  0",
                "4 白红卫  23  1"
            )
    
            val stuRDD:RDD[String] = sc.parallelize(stuList)
    
            val girls = stuRDD.filter(stu => {
    //            val fields = stu.split("\\s+")
    //            val gender = fields(3)
    //            gender == "0"
                stu.endsWith("0")
            })
            girls.foreach(println)
        }

### 1.4. sample ###

这是Spark中的抽样算子，从RDD中抽取一定比例的数据，接收三个参数：

withReplacement: Boolean, true有放回的抽样, false无放回的抽样

fraction: Double，样本空间大小，占总大小的比例的分数的表现形式，20%–>0.2

seed: Long ，抽样过程中的随机数种子。

注意：该抽样算子sample是非准确式的抽样，加入rdd中有1000个记录，抽取0.2的数据，按理结果应该有200个，真实结果可能会在200上下左右浮动。

作用就是，对整体空间做一个大概的预估，key/数据的分布情况，在解决数据倾斜的时候有着非常重要的角色。

def sampleOps(sc:SparkContext): Unit = {
            val list = 1 to 10000
            val listRDD = sc.parallelize(list)
            val sampledRDD = listRDD.sample(false, 0.2)
            println("抽取样本空间的大小：" + sampledRDD.count())
        }

### 1.5. union ###

在spark中的union操作，作用和DB中的union all的作用相同，不会去重。

union：返回两张表中去重之后的结果

union all：返回两张表的所有结果。

//作用类似于union all
        def unionOps(sc: SparkContext): Unit = {
            val listRDD1 = sc.parallelize(1 to 5)
            val listRDD2 = sc.parallelize(3 to 10)
            
            val unionRDD = listRDD1.union(listRDD2)
            unionRDD.foreach(println)
        }

### 1.6. join ###

表的关联操作，这里的表示泛化的概念，只要是数据集都可以理解为表。

join有哪些操作：

交叉连接：across join 写sql的时候有表的关联，但是没有on的连接字段，这会造成笛卡尔积。

select a.\* , b.\* from A a across join B b

内连接: inner（通常可省略） join，又被称为等值连接，返回左右表中都有的内容。

select a.\* , b.\* from A a inner join B b on [a.id][] = b.aid;

select a.\* , b.\* from A a, B b where [a.id][] = b.aid;

外连接 outer join 非等值连接

左外连接：left outer join 返回左表所有数据，右表没有显示为null

select a.\* , b.\* from A a left outer join B b on [a.id][] = b.aid;

右外连接: right outer join返回右表所有数据，左表没有显示为null

select a.\* , b.\* from A a right outer join B b on [a.id][] = b.aid;

半连接 semi join

全连接 full outer join 全连接： left outer join + right outer join

说明：经过内连接join操作之后的数据是确定的。外连接操作之的数据是确定的吗？不确定，有可能有，有可能没有-----Option。

/**
          * stu表
          *     id name age  class
          * 成绩表
          *     id sid  course  score
          *
          * 查询有成绩的学生的所有的信息-->inner
          *   select
                s.*, s1.*
              from stu s
              inner join score s1 on s.id = s1.sid
          * 查询所有学生的成绩
          *   select
          *     s.*, s1.*
          *   from stu s
          *   left join s1 on on s.id = s1.sid
          * 所有的join操作，必须要求的RDD的类型时<K, V>
          * K就是关联字段
          */
        def joinOps(sc: SparkContext): Unit = {
            val stu = List(
                "1 刘梦男  22  bd-1901-bj",
                "2 常国龙  25  bd-1901-bj",
                "3 张湟熹  24  bd-1901-sz",
                "4 胡盼盼(男) 18  bd-1901-wh"
            )
    
            val scores = List(
                "1 1 math 82",
                "2 1 english 0",
                "3 2 chinese 85.5",
                "4 3 PE 99",
                "5 10 math 99"
            )
            val stuRDD:RDD[String] = sc.parallelize(stu)
            val scoreRDD:RDD[String] = sc.parallelize(scores)
            //查询所有的学生信息
            val sid2StuInfoRDD:RDD[(String, String)] = stuRDD.map(stuLine => {
                val sid = stuLine.substring(0, 1)
                val stuInfo = stuLine.substring(1).trim
                (sid, stuInfo)
            })
    
            val sid2ScoreInfo:RDD[(String, String)] = scoreRDD.map(scoreLine => {
                val fields = scoreLine.split("\\s+")
                val sid = fields(1)
                val scoreInfo = fields(2) + " " + fields(3)
                (sid, scoreInfo)
            })
            //(K, V) join (K, W) ---> (K, (V, W))
            val joinedRDD:RDD[(String, (String, String))] = sid2StuInfoRDD.join(sid2ScoreInfo)
            /*joinedRDD.foreach(t => {
                println(s"sid:${t._1}\tstuInfo:${t._2._1}\tscoreInfo:${t._2._2}")
            })*/
            joinedRDD.foreach{case (sid, (stuInfo, scoreInfo)) => {
                println(s"sid:${sid}\tstuInfo:${stuInfo}\tscoreInfo:${scoreInfo}")
            }}
            println("------------------------------")
            val leftJoined:RDD[(String, (String, Option[String]))] = sid2StuInfoRDD.leftOuterJoin(sid2ScoreInfo)
            leftJoined.foreach{case (sid, (stuInfo, option)) => {
                println(s"sid:${sid}\tstuInfo:${stuInfo}\tscoreInfo:${option.getOrElse(null)}")
            }}
            println("--------------full join----------------")
    
            val fullJoinedRDD:RDD[(String, (Option[String], Option[String]))] = sid2StuInfoRDD.fullOuterJoin(sid2ScoreInfo)
    
            fullJoinedRDD.foreach{case (sid, (stuOption, scoreOption)) => {
                println(s"sid:${sid}\tstuInfo:${stuOption.getOrElse(null)}\tscoreInfo:${scoreOption.getOrElse(null)}")
            }}
    
        }

### 1.7. groupByKey ###

对rdd中的数据按照key进行分组，首先必须要有key，rdd的数据类型就必须是(K, V),经过分组之后相同的key的数据拉取到一起，组成了一个集合。

(K, V).groupByKey() --> (K, Iterable\[V\])

不建议使用这个groupByKey，可以使用reduceByKey或者aggregateByKey来代替gbk。因为性能差，没有本地shuffle==combiner。

def gbkOps(sc: SparkContext): Unit = {
            val stu = List(
                "1 刘梦男  22  bd-1901-bj",
                "2 常国龙  25  bd-1901-bj",
                "3 张湟熹  24  bd-1901-sz",
                "4 胡盼盼(男) 18  bd-1901-wh",
                "5 周联动 18  bd-1901-wh",
                "4 张震 18  bd-1901-sz"
            )
    
            //按照班级进行分组 -->sql中的groupBy
            val stuRDD:RDD[String] = sc.parallelize(stu)
            val class2Info:RDD[(String, String)] = stuRDD.map(stuLine => {
                val clazz = stuLine.substring(stuLine.indexOf("bd"))
                val info = stuLine.substring(0, stuLine.indexOf("bd")).trim
                (clazz, info)
            })
    
            val gbkRDD:RDD[(String, Iterable[String])] = class2Info.groupByKey()
    
    
            gbkRDD.foreach{case (clazz, infos) => {
                /*for(info <- infos) {
                    println(s"${clazz} ---> ${info}")
                }*/
                println(s"${clazz} ---> ${infos}")
            }}
            println("-----------------------------------------")
            //ClassTag是类型的标记接口
            val gbRDD:RDD[(String, Iterable[String])] = stuRDD.groupBy(stuLine => stuLine.substring(stuLine.indexOf("bd")))(
                ClassTag.Object.asInstanceOf[ClassTag[String]]
            )
            gbRDD.foreach{case (clazz, infos) => {
                println(s"${clazz} ===> ${infos}")
            }}
        }

### 1.8. reduceByKey ###

按照key执行reduce操作，也要求数据类型(K, V) —> (K, V)

国际惯例的案例：

def rbkOps(sc: SparkContext): Unit = {
            val list = List(
                "yan xin     xin",
                "zhai zhao pin"
            )
    
            val listRDD = sc.parallelize(list)
    
            val wordsRDD = listRDD.flatMap(line => line.split("\\s+"))
            val pairsRDD:RDD[(String, Int)]= wordsRDD.map((_, 1))
            val rbkRDD:RDD[(String, Int)] = pairsRDD.reduceByKey(_+_)
            rbkRDD.foreach(println)
        }

### 1.9. distinct ###

去重，及时sql中的关键字distinct。

def distinctOps(sc:SparkContext): Unit = {
            val list = List(
                "蔡金廷",
                "蔡金廷",
                "蔡金廷",
                "蔡金廷"
            )
    
            val listRDD:RDD[String] = sc.parallelize(list)
            //numPartitions:在去重的时候指定几个分区-->task,也就是说同时运行几个task来进行去重
            listRDD.distinct(4).foreach(println)
        }

### 1.10 combineByKey \*\*\* ###

我们刚才通过查看源码，发现reduceByKey和groupByKey底层都是通过combineByKeyWithClassTag来实现的，有一个简写方式combineByKey。

意为按照key进行聚合combine操作，聚合操作有很多，groupBy、reduceBy、count等等。

自定义的combineByKey。通过学习combineBykey模拟reduceByKey和groupByKey来理解什么是分布式计算。

1.  combineByKey模拟reduceByKey
    
        object _02CombineByKey2ReduceByKeyOps {
            def main(args: Array[String]): Unit = {
                val conf = new SparkConf()
                            .setAppName("cbk2rbk")
                            .setMaster("local[*]")
                val sc = new SparkContext(conf)
        
        
        
                val list = List(
                    "hello you hello me",
                    "hello you shit me",
                    "hello you oh shit",
                    "me you hello me"
                )
        
                val listRDD = sc.parallelize(list)
        
                val pairsRDD:RDD[(String, Int)] = listRDD.flatMap(_.split("\\s+")).map((_, 1))
        
                //reduceByKey
                println("-----------传统的rbk操作--------------")
                pairsRDD.reduceByKey(_+_).foreach(println)
                println("-----------combineByKey模拟rbk操作--------------")
                pairsRDD.combineByKey(createCombiner, mergeValue, mergeCombiners)
                            .foreach(println)
        
        
                sc.stop()
            }
        
            /**求：1 + 10
              * var sum1 = 1
              *
              * for(i <- 2 to 10) {
              *     sum1 = sum1 + i
              * }
              * var sum2 = 0
              * for(i <- 1 to 10) {
              *     sum2 = sum2 + i
              * }
              *  初始化聚合的结果类型
              */
            def createCombiner(value:Int):Int = {
                value
            }
            //分区内的聚合操作（map端的操作）
            def mergeValue(sum:Int, value:Int):Int = {
                sum + value
            }
            //分区间的聚合操作(reduce端的操作)
            def mergeCombiners(sum1:Int, sum2:Int): Int = {
                sum1 + sum2
            }
        }
2.  combineByKey模拟groupByKey
    
        object _03CombineByKey2GroupByKeyOps {
            def main(args: Array[String]): Unit = {
                val conf = new SparkConf()
                            .setAppName("cbk2gbk")
                            .setMaster("local[2]")//thread -- task--->partition(block)
                val sc = new SparkContext(conf)
        
                val stu = List(
                    "1 刘梦男  22  bd-1901-bj",
                    "2 常国龙  25  bd-1901-bj",
                    "8 汪锦奇  24  bd-1901-sz",
                    "3 张湟熹  24  bd-1901-sz",
                    "4 胡盼盼(男) 18  bd-1901-wh",
                    "6 陈抒学  25  bd-1901-bj",
                    "7 曹煜  25  bd-1901-bj",
                    "5 周联动 18  bd-1901-wh",
                    "4 张震 18  bd-1901-sz"
                )
        
                //按照班级进行分组 -->sql中的groupBy
                val stuRDD:RDD[String] = sc.parallelize(stu)
                val class2Info:RDD[(String, String)] = stuRDD.map(stuLine => {
                    val clazz = stuLine.substring(stuLine.indexOf("bd"))
                    (clazz, stuLine)
                })
                class2Info.saveAsTextFile("file:///E:/data/out/cbk")
                val gbkRDD:RDD[(String, Iterable[String])] = class2Info.groupByKey()
        
                println("-----------传统的gbk操作--------------")
                gbkRDD.foreach{case (clazz, infos) => {
                    println(s"${clazz} ---> ${infos}")
                }}
                println("-----------combineByKey模拟gbk操作--------------")
        
                class2Info.combineByKey(
                        (info:String) => createCombiner(info),
                        (buffer:ArrayBuffer[String], info:String) => mergeValue(buffer, info),
                        (buffer1:ArrayBuffer[String], buffer2:ArrayBuffer[String]) => mergeCombiners(buffer1, buffer2),
                        new MyPartitioner(3))//该分区动作不是combineByKey数据输入进来是对数据做分区，经过combineByKey计算完毕之后将结果做分区
                   .saveAsTextFile("file:///E:/data/out/cbk1")   
                sc.stop()
            }
            /*
                初始化操作，确定聚合操作之后的结果类型
                在每一个分区内相同的key，需要调用一次该操作，并将其中的一个元素用于初始化操作
             */
            def createCombiner(info:String):ArrayBuffer[String] = {
                println("----createCombiner------>info: " + info)
                val ab = ArrayBuffer[String]()
                ab.append(info)
                ab
            }
            /**
              * 分区内的相同key的聚合操作
              *
              */
            def mergeValue(ab:ArrayBuffer[String], info:String):ArrayBuffer[String] = {
                println(s"----mergeValue------>ab:${ab.mkString(",")}《===》info: ${info}")
                ab.append(info)
                ab
            }
            /**
              * 分区间的相同key的聚合操作
              */
            def mergeCombiners(ab1:ArrayBuffer[String], ab2:ArrayBuffer[String]):ArrayBuffer[String] = {
                println(s"----mergeCombiners------>ab1:${ab1.mkString(",")}《===》ab2:${ab2.mkString(",")}")
                ab1 ++ ab2
            }
        }
        class MyPartitioner(partitions:Int) extends Partitioner {
            override def numPartitions = partitions
            override def getPartition(key: Any) = {
                val ret = key match {
                    case clazz:String => {
                        //bd-1901-wh
                        val city = clazz.substring(clazz.lastIndexOf("-") + 1)
                        city match {
                            case "wh" => 0
                            case "bj" => 1
                            case "sz" => 2
                        }
                    }
                    case _ => 0
                }
                ret
            }
        }

### 1.11. aggregateByKey \*\*\* ###

aggregateBykey本质和combineByKey一样的。使用方面，也就是参数传递的方面略有差异。

1.  aggregateBykey模拟groupByKey
    
        object _04AggregateByKey2GroupByKeyOps {
            def main(args: Array[String]): Unit = {
                val conf = new SparkConf()
                    .setAppName("abk2gbk")
                    .setMaster("local[2]")
                //thread -- task--->partition(block)
                val sc = new SparkContext(conf)
        
                val stu = List(
                    "1 刘梦男  22  bd-1901-bj",
                    "2 常国龙  25  bd-1901-bj",
                    "8 汪锦奇  24  bd-1901-sz",
                    "3 张湟熹  24  bd-1901-sz",
                    "4 胡盼盼(男) 18  bd-1901-wh",
                    "6 陈抒学  25  bd-1901-bj",
                    "7 曹煜  25  bd-1901-bj",
                    "5 周联动 18  bd-1901-wh",
                    "4 张震 18  bd-1901-sz"
                )
        
                //按照班级进行分组 -->sql中的groupBy
                val stuRDD: RDD[String] = sc.parallelize(stu)
                val class2Info: RDD[(String, String)] = stuRDD.map(stuLine => {
                    val clazz = stuLine.substring(stuLine.indexOf("bd"))
                    (clazz, stuLine)
                })
                val gbkRDD: RDD[(String, Iterable[String])] = class2Info.groupByKey()
        
                println("-----------传统的gbk操作--------------")
                gbkRDD.foreach { case (clazz, infos) => {
                    println(s"${clazz} ---> ${infos}")
                }
                }
                println("-----------aggregateByKey模拟gbk操作--------------")
        
                val abk = class2Info.aggregateByKey(ArrayBuffer[String]())(seqOp, combOp)
                abk.foreach(println)
            }
        
            /**
              * var sum = 0
              * for(i <- 1 to 10) {
              *     sum = sum + i
              * }
              * var sum1 = 1
              * for(i <- 2 to 10) {
              *     sum = sum + i
              * }
              */
            def seqOp(ab:ArrayBuffer[String], info:String): ArrayBuffer[String] = {
                ab.append(info)
                ab
            }
        
            def combOp(ab1:ArrayBuffer[String], ab2:ArrayBuffer[String]): ArrayBuffer[String] = {
                ab1 ++ ab2
            }
        }
2.  aggregateBykey模拟reduceByKey

object _05AggregateByKey2ReduceByKeyOps {
        def main(args: Array[String]): Unit = {
            val conf = new SparkConf()
                        .setAppName("abk2rbk")
                        .setMaster("local[*]")
            val sc = new SparkContext(conf)
    
            val list = List(
                "hello you hello me",
                "hello you shit me",
                "hello you oh shit",
                "me you hello me"
            )
    
            val listRDD = sc.parallelize(list)
    
            val pairsRDD:RDD[(String, Int)] = listRDD.flatMap(_.split("\\s+")).map((_, 1))
    
            //reduceByKey
            println("-----------传统的rbk操作--------------")
            pairsRDD.reduceByKey(_+_).foreach(println)
            println("-----------combineByKey模拟rbk操作--------------")
    
            pairsRDD.aggregateByKey(0)(seqOp, combOp).foreach(println)
    
            sc.stop()
        }
        /**
          * var sum = 0
          * for(i <- 1 to 10) {
          *     sum = sum + i
          * }
          * var sum1 = 1
          * for(i <- 2 to 10) {
          *     sum = sum + i
          * }
          */
        def seqOp(sum:Int, i:Int): Int = {
            sum + i
        }
    
        def combOp(sum1:Int, sum2:Int): Int = {
            sum1 + sum2
        }
    }

combineByKey和aggregateByKey的区别？

1、本质上二者是没有区别的，都是combineByKeyWithClassTag的简写方式，都是进行聚合操作。

2、combineByKey接收三个聚合参数，会将每个分区中相同key的第一个值用于初始化，而aggregateByKey接收一个0值。所以在分区内进行聚合操作的时候，combineByKey会被调用n-1次方法，aggregateByKey会被调用n次，其中n就是一个分区内相同key的数据量。

3、aggregateByKey其实更加适合处理那些聚合前后类型都一致的操作，combineByKey比较适合处理那些聚合前后类型不一致的操作。

pairsRDD.aggregateByKey(0)(_+_, _+_).foreach(println)

pairsRDD.combineByKey((num:Int) => num, (sum1:Int, num:Int) => sum1 + num, (sum:Int, sum1:Int) => sum + sum1)

## 1.2 Action算子 ##

action操作，是行动算子，是触发spark作业执行的动因。

### 2.1. foreach ###

略

### 2.2. collect ###

是将集群中的数据(rdd对应partition的数据)拉取到dirver(本地)，所以我们需要非常**慎重**的使用这个算子，避免出现OOM的异常，所以一般要拉去的话我们要么在拉取之前进行filter；要么使用下面的take算子来获取对应的数据。

### 2.3. count ###

略

### 2.4. saveAsTextFile/saveAsHadoopFile/saveAsHadoopAPIFile ###

/*
                saveAsTextFile
                    将rdd以普通文本的方式进行存储，存储的格式()
             */
    //        retRDD.saveAsTextFile("file:///E:/data/out/astf")
            /*
                saveAsSequenceFile
                    将rdd以sequenceFile的格式存储
             */
    //        retRDD.saveAsSequenceFile("file:///E:/data/out/sequ", Option(classOf[DefaultCodec]))
            /*
                saveAsHadoopFile
                saveAsNewHadoopAPIFile
                    都是使用的hadoop的格式化输出OutputFormat 这两个方法的区别主要就是OutputFormat的区别
                    saveAsHadoopFile --> org.apache.hadoop.mapred.OutputFormat
                    saveAsNewHadoopAPIFile --> org.apache.hadoop.mapreduce.OutputFormat
             */
            retRDD.saveAsNewAPIHadoopFile("file:///E:/data/out/hadoop",
                    classOf[Text],
                    classOf[IntWritable],
                    classOf[TextOutputFormat[Text, IntWritable]]
                )

### 2.5. take(n)/frist/takeOrdered ###

take获取集合中的前N个值，take(1)=first,如果说这个集合是有序的，take(n)—>TopN

println("-------take(2)-------------")
    retRDD.take(2).foreach(println)
    println("-------first-------------")
    println(retRDD.first())
    println("--------takeOrdered----------------------")
    implicit val order = new Ordering[(String, Int)](){
        override def compare(x: (String, Int), y: (String, Int)) = {
            var ret = y._2.compareTo(x._2)
            if(ret == 0) {
                ret = y._1.compareTo(x._1)
            }
            ret
        }
    }
    retRDD.takeOrdered(5).foreach(println)

### 2.6. countByKey ###

countByKey/
            //通过每个key出现的次数
            val k2counts = pairsRDD.countByKey()
            for((k, v) <- k2counts) {
                println(s"${k}---$v")
            }
            //使用groupByKey、reduceByKey、countByKey   --->wordcount
            println("-----------------------------------")
            val gbks = pairsRDD.groupByKey()
    //        gbks.foreach(println)
            gbks.map{case (key, values) => (key, values.size)}.foreach(println)

### 2.7. reduce ###

reduceByKey是一个transformation，reduce是一个action。

val sum = pairsRDD.values.reduce(_+_)
    println(sum)

[a.id]: http://a.id