Spark算子代码汇总

灰太狼 2021-11-26 20:40 450阅读 0赞

# Spark算子代码汇总 #

算子分类

Transformations转换算子延迟执行 需要Action算子触发才执行

Action行动算子，触发执行

控制算子 cache persist checkpoint

### Transformations ###

延迟执行 需要Action算子触发才执行

#### map- 1:1 取数据数据是一行 ####

#### flatMap-1:n 数据是多行拆分过的 ####

#### filter-过滤 ####

#### sortByKey-排序 ####

#### sortBy-排序 按照v(Int)进行排序,默认正序排列 ####

#### sample-随机抽样 ####

//map 1:1 数据是一行
    //    lines.map(one=>{
    //      one+"#"
    //    }).foreach(println)
    
        //flatMap 1:n 数据是多行拆分过的
    //    lines.flatMap(one=>{one.split(" ")}).foreach(println)
    
        //过滤-filter
    //    val rdd1: RDD[String] = lines.flatMap(one=>{one.split(" ")})
    //    rdd1.filter(one=>{
    //      "book".equals(one)
    //    }).foreach(println)
    
        //排序-ortByKey
    //    val words: RDD[String] = lines.flatMap(one=>{one.split(" ")})
    //    val pairWords: RDD[(String, Int)] = words.map(one=>{(one,1)})
    //    val reduceResult: RDD[(String, Int)] = pairWords.reduceByKey((v1:Int,v2:Int)=>{v1+v2})
    //    //按照k(String)进行v的逆序
    //    //从map中取出数据并翻转，此时是数据在前单词在后
    //    val teransRDD: RDD[(Int, String)] = reduceResult.map(tp=>{tp.swap})
    //    val result: RDD[(Int, String)] = teransRDD.sortByKey(false)
    //    //再次转换kv位置
    //    result.map(_.swap).foreach(println)
    
        //排序-sortBy
        //按照v(Int)进行排序，不写false，默认正序排列
    //    reduceResult.sortBy(tp=>{tp._2},false)foreach(println)
    
        //随机抽样-sample
        //(抽到所需数据后是继续，抽样占总数的比率，抽样的随机数据会固定)
    //    val result: RDD[String] = lines.sample(true,0.1,100L)
    //    result.foreach(println)

#### join ####

不相同的元素不打印

val nameRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",20),("wangwu",22),("maliu",24)))
        val soreRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",100),("lisi",200),("wangwu",300),("tianqi",400)))
    
        //jion
        val result: RDD[(String, (Int, Int))] = nameRDD.join(soreRDD)
        result.foreach(println)

#### leftOuterJoin-偏左合并 ####

leftOuterJoin 以左为主，左边的全都打印，没有的显示None

左边少，右边多的元素，不打印

遍历元素可以取元组下标

val result: RDD[(String, (Int, Option[Int]))] = nameRDD.leftOuterJoin(soreRDD)
    result.foreach(one=>{
      val name = one._1
      val age = one._2._1
      val sore = one._2._2.getOrElse("666")
      println(s"name = $name, age=$age, sore = $sore")
    })

#### rightOuterJoin-偏右合并 ####

用法等同于leftOuterJoin

#### fullOuterJoin-全合并，None补空 ####

左右两边都打印，没有的元素用None表示

val result: RDD[(String, (Option[Int], Option[Int]))] = nameRDD.fullOuterJoin(soreRDD)
    result.foreach(println)

#### getNumPartitions-分区数 ####

//      //打印分区数
        val p1 = nameRDD.getNumPartitions
        val p2 = soreRDD.getNumPartitions
        println(p1,p2)
          //join数据后，新的RDD分区数与诸多父RDD的最大分区数保持一致
        val result: RDD[(String, (Int, Int))] = nameRDD.join(soreRDD)
        println(result.getNumPartitions)

#### intersection-取交集 ####

val rdd1: RDD[Int] = sc.parallelize(List[(Int)](1,2,3))
    val rdd2: RDD[Int] = sc.parallelize(List[(Int)](2,3,6))
    
    //intersection取交集
    val result: RDD[Int] = rdd1.intersection(rdd2)

#### subtract-取差集 ####

val rdd1: RDD[Int] = sc.parallelize(List[(Int)](1,2,3))
    val rdd2: RDD[Int] = sc.parallelize(List[(Int)](2,3,6))
    
    //subtract取差集
    val result: RDD[Int] = rdd1.subtract(rdd2)
    result.foreach(println)

#### map 和 partition ####

//N次执行加载N次

val rdd: RDD[String] = sc.parallelize(List[(String)]("hello1","hello2","hello3","hello4","hello5","hello6"),2)
    
    //建立6次连接-map
    rdd.map(one=>{
      println("创建数据库连接...")
      println(s"加载数据 $one")
      println("创建数据库连接...")
      one+ "!!!"
    }).count()
    
    //建立6次连接-foreach  不需要返回数据
      rdd.foreach(one=>{
      println("创建数据库连接...")
      println(s"加载数据 $one")
      println("创建数据库连接...")
    })

#### mapPartitions 和 foreachPartition ####

执行N次，加载分区次数

//建立2次连接-map，上面分区数=2  业务需要对数据再进行处理使用mapPartition
      val unit: RDD[String] = rdd.mapPartitions(iter => {
        val list = new ListBuffer[String]
        println("建立数据库连接。。。")
        while (iter.hasNext) {
          val str: String = iter.next()
          println(s"插入数据 $str")
          list.+=(str)
        }
        println("关闭数据库连接。。。")
        list.iterator
      })
      unit.count()
    
     // 建立2次连接-map，上面分区数=2  不需要返回数据 当业务插入数据库结束情况下考虑使用
      rdd.foreachPartition(iter => {
      val list = new ListBuffer[String]
      println("建立数据库连接。。。")
      while (iter.hasNext) {
        val str: String = iter.next()
        println(s"插入数据 $str")
      }
      println("关闭数据库连接。。。")
    })

#### distinct-去重 ####

val rdd: RDD[String] = sc.parallelize(List[(String)]("a","b","a","c","d","d"))
    rdd.map(one=>{(one,1)}).reduceByKey((v1,v2)=>{v1+v2}).map(one=>{one._1}).foreach(println)
    val unit: RDD[String] = rdd.distinct()
    unit.foreach(print)

#### cogroup-合并 ####

val nameRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("zhangsan",180),("zhangsan",180),("lisi",20),("wangwu",22),("maliu",24)),3)
    val soreRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",100),("lisi",200),("wangwu",300),("tianqi",400),("tianqi",40000)),4)
    val unit: RDD[(String, (Iterable[Int], Iterable[Int]))] = nameRDD.cogroup(soreRDD)
    unit.foreach(println)
    println(unit.getNumPartitions)

#### mapPartitionWithIndex-取分区索引号和该分区中的数据 ####

val rdd1: RDD[String] = sc.parallelize(List[String](
          "love1", "love2", "love3", "love4",
          "love5", "love6", "love7", "love8",
          "love9", "love10", "love11", "love12"
        ), 3)
    
        /**
          * mapPartitionsWithIndex(index,iter)
          * 拿到取分区索引号和该分区中的数据
          */
        //
        val rdd2: RDD[String] = rdd1.mapPartitionsWithIndex((index, iter) => {
          val list = new ListBuffer[String]()
          while (iter.hasNext) {
            val one = iter.next()
            list.+=(s"【$index】【$one】")
          }
          list.iterator
        })
    //    rdd2.foreach(println)

#### repartition-增加分区数 ####

/**
          * repartition
          * 控制RDD的分区数目，增减
          * 宽依赖，会产生suffle
          * 常用于增加分区
          */
        val  rdd3 = rdd2.repartition(2)
        val rdd4: RDD[String] = rdd3.mapPartitionsWithIndex((index, iter) => {
          val list = new ListBuffer[String]()
          while (iter.hasNext) {
            val one = iter.next()
            list.+=(s"【$index】【$one】")
          }
          list.iterator
        })
    //    val result = rdd4.collect().foreach(println)

#### coalesce-减少分区（常用） ####

/**
          * coalesce
          * 常用于减少分区
          * 控制RDD的分区数目，增减 窄依赖
          * 添加属性true  = repartition 变成宽依赖差生shuffle
          * 当coalesce 由少的分区分到多的分区时，在不让产生shuffle情况下，是没有作用的
          */
          val rdd5 = rdd2.coalesce(2)
    //    val  rdd5 = rdd2.coalesce(2,true)
        val rdd6: RDD[String] = rdd5.mapPartitionsWithIndex((index, iter) => {
          val list = new ListBuffer[String]()
          while (iter.hasNext) {
            val one = iter.next()
            list.+=(s"【$index】【$one】")
          }
          list.iterator
        })
        val result = rdd6.collect().foreach(println)

#### groupByKey-相同K放一起 ####

/**
      * groupByKey
      * 相同k数据放一起
      * (k,CompactBuffer(v1, v2))
      */
    val rdd1 = sc.parallelize(List[(String,Int)](("a",1),("a",2),("b",333),("b",444)))
    rdd1.groupByKey().foreach(println)

#### zip-1:压缩集合元素 ####

val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan", "lisi", "wangwu", "maliu"))
    val rdd2: RDD[String] = sc.parallelize(List[String]("100", "200", "300", "400"))
    /**
      * zip
      * 1:!压缩元素，2集合元素个数必须一样
      */
    val result: RDD[(String, String)] = rdd1.zip(rdd2)
    result.foreach(println)

#### zipWithIndex-压缩数据和所在索引 ####

/**
      * zipWithIndex
      * 压缩集合元素和该元素的索引
      */
     val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan", "lisi", "wangwu", "maliu"))
    val result1: RDD[(String, Long)] = rdd1.zipWithIndex()
    result1.foreach(println)

### Action行动算子 ###

触发执行

#### collect()-以数组方式回收数据到客户端 ####

#### foreach()-内外数据交互 ####

N个加载N次

#### foreachPartition-内外交互 ####

N个加载分区次数

// 建立2次连接-map，上面分区数=2  不需要返回数据 当业务插入数据库结束情况下考虑使用
      rdd.foreachPartition(iter => {
      val list = new ListBuffer[String]
      println("建立数据库连接。。。")
      while (iter.hasNext) {
        val str: String = iter.next()
        println(s"插入数据 $str")
      }
      println("关闭数据库连接。。。")
    })

#### take(n)-拿取数据的前n条 ####

#### first-拿取数据第一条 ####

#### count-计数数据 ####

//拿取数据的前5条
        val result: Array[String] = lines.take(3)
        result.foreach(println)
    
        //拿取数据第一条
        val result1: String = lines.first()
        println(result1)
    
        //计数数据
        val l = lines.count()
        println(l)
    
        //将计算结果回收到Driver端，在大数据情况中不建议使用
    //    val result2: Array[String] = lines.collect()
    //    result2.foreach(println)

#### countByKey-统计(k,n) ####

必须作用在KV格式的数据上

val rdd1 = sc.parallelize(List[(String,Int)](
          ("a",100),
          ("a",100),
          ("a",200),
          ("b",2),
          ("c",300),
          ("d",100),
          ("d",500)
        ))
        val stringToLong: collection.Map[String, Long] = rdd1.countByKey()

#### countByValue-统计((k,v)n) ####

不必作用在kv格式上

将原集合的每个kv看做每一个整体，再统计其整体出现的次数

val rdd2 = sc.parallelize(List[(String,Int)](
          ("a",100),
          ("a",100),
          ("a",200),
          ("b",2),
          ("c",300),
          ("d",100),
          ("d",500)
        ))
        val tupleToLong: collection.Map[(String, Int), Long] = rdd2.countByValue()
        tupleToLong.foreach(println)
        
        
    
        val rdd3 = sc.parallelize(List[Int](1,2,3,4,4,4,5,2))
        val intToLong: collection.Map[Int, Long] = rdd3.countByValue()
        intToLong.foreach(println)

#### reduce-集合所有元素简单计算 ####

/**
      * reduce
      * 值相加
      */
    val rdd = sc.parallelize(List[Int](1,2,3,4,5))
    //    val result: Int = rdd.reduce(_+_)
    val result: Int = rdd.reduce((v1, v2)=>{v1+v2})
    println(result)

#### union-单纯合并 ####

本质上数据的合并，打印的是两个RDD的数据，分区数=诸多父类分区数的和

val result: RDD[(String, Int)] = nameRDD.union(soreRDD)
    val partitions: Int = result.getNumPartitions
    result.foreach(println)
    println(partitions)

### 控制算子-持久化 ###

#### 控制算子 -cache ####

默认将RDD的数据持久化到内存中，cache是懒执行，需要触发算子触发

注意：chche () = persist()=persist(StorageLevel.Memory\_Only)

//测试数据200M+ 时间3331	85
    
    var rdd: RDD[String] = sc.textFile("./Test_spark/data/data.txt")
    //懒执行，需要触发，第一次加载出发后放入内存
    rdd = rdd.cache()
    val startTime1 = System.currentTimeMillis()
    val result1 = rdd.count()
    val endTime1 = System.currentTimeMillis()
    println(s"条数=$result1, 时间=${endTime1-startTime1}")
    
    
    val startTime2 = System.currentTimeMillis()
    val result2 = rdd.count()
    val endTime2 = System.currentTimeMillis()
    println(s"条数=$result2, 时间=${endTime2-startTime2}")

#### countByKey-统计(k,n) ####

必须作用在KV格式的数据上

#### countByValue-统计((k,v)n) ####

不必作用在kv格式上

将原集合的每个kv看做每一个整体，再统计其整体出现的次数

#### reduce-集合值相加 ####

#### 控制算子 -persist ####

可以指定持久化的级别。最常用的是MEMORY\_ONLY和MEMORY\_AND\_DISK。

”\_2”表示有副本数。

常用的级别：  
 MEMORY\_ONLY -最常用，不需要序列化  
 MEMORY\_ONLY\_SER 序列化后放在内存  
 MEMORY\_AND\_DISK 内存放不下，放内存和磁盘  
 MEMORY\_AND\_DISK\_SER 序列化后放入磁盘或内存

#### 注意： ####

1、cache,persist,checkpoint都是懒执行，最小化单位是partition

2、cache和persist之后都可以直接赋值给一个值(rdd = rdd.cache() / rdd.cache()),下次直接使用这个值，就是使用持久化的数据

3、但是后面不能紧跟action算子 . xxxx 如果紧跟Action算子，则后面再使用的话就是固定的值，不是持久化的数据

4、cache和persist的持久化数据，当application执行完成之后会自动清除

#### 控制算子 -checkpoint ####

会创建持久化数据的目录，当RDD的lineage某些逻辑非常复杂，还可以使用checkpoint对RDD进行持久化  
不要经常对RDD使用checkpoint，这样会导致application执行缓慢，因为使用checkpoint相当于从磁盘中读取数据  
checkpoint执行流程  
 1.当spark执行完成之后，Spark框架会从后往前回溯，并找到chekpointRDD进行标记  
 2.回溯完成之后，Spark会重新启动Job，查看并计算chekpointRDDRDD的数据，将该RDD的数据存放在指定的持久化目录中  
 3.切断RDD之间的依赖关系  
 优化：持久化之前最好先cache下