【RDD Transformation】leftOuterJoin、rightOuterJoin、cartesian、groupBy

古城微笑少年丶 2022-11-25 13:13 31阅读 0赞

# 一、Rdd转化算子 #

1、【leftOuterJoin】左连接：以左边rdd为基准，若连接rdd没有左边rdd中的元素，则给null；

2、【rightOuterJoin】右连接：以右边rdd为基准，若连接rdd没有右边rdd中的元素，则给null；

3、【cartesian】笛卡儿积；

4、【groupBy】groupBy分组：根据传入的参数进行分组，相同参数是一组；

## 二、实例 ##

package com.cn.rddOperator
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    import scala.collection.mutable.{ArrayBuffer, ListBuffer}
    
    object Transformation03 {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("transformation03").setMaster("local[1]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val rdd01=sc.parallelize(List(("tom",1),("jack",3),("dack",5)))
        val rdd02=sc.parallelize(List(("jack",3),("tom",7),("pig",2)))
    
        /**
          * 左连接：以左边rdd为基准，若连接rdd没有左边rdd中的元素，则给null;
          */
        val rdd03: RDD[(String, (Int, Option[Int]))] = rdd01.leftOuterJoin(rdd02)
        println(rdd03.collect().toBuffer)//ArrayBuffer((tom,(1,Some(7))), (jack,(3,Some(3))), (dack,(5,None)))
    
        /**
          * 右连接：以右边rdd为基准，若连接rdd没有右边rdd中的元素，则给null;
          */
        val rdd04: RDD[(String, (Option[Int], Int))] = rdd01.rightOuterJoin(rdd02)
        println(rdd04.collect().toBuffer)//ArrayBuffer((tom,(Some(1),7)), (pig,(None,2)), (jack,(Some(3),3)))
    
        /**
          * 笛卡儿积
          */
        val rdd05: RDD[((String, Int), (String, Int))] =rdd01 cartesian rdd02
        println(rdd05.collect().toBuffer)//ArrayBuffer(((tom,1),(jack,3)), ((tom,1),(tom,7)), ((tom,1),(pig,2)), ((jack,3),(jack,3)), ((jack,3),(tom,7)), ((jack,3),(pig,2)), ((dack,5),(jack,3)), ((dack,5),(tom,7)), ((dack,5),(pig,2)))
    
        /**
          * groupBy分组：根据传入的参数进行分组，相同参数是一组
          */
        val rdd06: RDD[(String, Int)] = sc.parallelize(List(("dog",2),("pig",3),("dog",5),("dack",6),("fox",8),("pig",9)))
        val rdd07: RDD[(String, Iterable[(String, Int)])] = rdd06.groupBy(_._1)
        println(rdd07.collect().toBuffer)//ArrayBuffer((fox,CompactBuffer((fox,8))), (dog,CompactBuffer((dog,2), (dog,5))), (pig,CompactBuffer((pig,3), (pig,9))), (dack,CompactBuffer((dack,6))))
      }
    }