Spark算子总结

矫情吗；* 2022-05-29 03:45 300阅读 0赞

# Spark算子总结 #

spark 算子分为两类：transformation和Action

## 1.常用spark action ##

> val rdd1 = sc.parallelize(List(1,2,3,4,5), 2)

--------------------

### collect 将rdd转化成数组（Array） ###

> rdd1.collect

--------------------

### reduce 按照一定的方法将元素进行合并 ###

> val rdd2 = rdd1.reduce(\_+\_)

--------------------

### count 统计元素数量 ###

> rdd1.count

--------------------

### top ###

取最大的n个

> rdd1.top(2)

## 对数据集进行排序，然后取出最大的两个 ##

### take ###

取出前i个元素，不排序

> rdd1.take(2)

--------------------

### first(similer to take(1)) ###

> rdd1.first

--------------------

### takeOrdered ###

> rdd1.takeOrdered(3)

--------------------

### checkpoint ###

由于计算过程是在内存进行，然后spill出来，每到达一个checkpoint就会将内存中的数据写入到磁盘，这个功能就是手动使其到达checkpoint

sc.setCheckpointDir("hdfs://bigdata01:9000/ck")
    val rdd = sc.textFile("hdfs://bigdata01:9000/wc").flatMap(_.split("")).map((_, 1)).reduceByKey(_+_)
    rdd.checkpoint
    rdd.isCheckpointed
    rdd.count
    rdd.isCheckpointed
    rdd.getCheckpointFile

--------------------

## 2.常用 spark Transformation(即转换，延迟加载) ##

### 通过并行化scala集合创建RDD（弹性分布式数据集） ###

里面可以接受Array和List

> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))

--------------------

#### 查看该rdd的分区数量 ####

> rdd1.partitions.length

--------------------

### union求并集，注意类型要一致 ###

> def union(other: RDD\[T\]): RDD\[T\]

val rdd6 = sc.parallelize(List(5,6,4,7))
    val rdd7 = sc.parallelize(List(1,2,3,4))
    val rdd8 = rdd6.union(rdd7)
    rdd8.distinct.sortBy(x=>x).collect

--------------------

### intersection ###

> def intersection(other: RDD\[T\], numPartitions: Int): RDD\[T\]  
> def intersection(other: RDD\[T\], partitioner: Partitioner)(implicit ord: Ordering\[T\] = null): RDD\[T\]  
> def intersection(other: RDD\[T\]): RDD\[T\]

求求出两个rdd的交集，有三个函数签名

> val rdd9 = rdd6.intersection(rdd7)

## 会输出只含有4的rdd ##

### join 将数据集连接聚合，有点类似数据库里面的join ###

必须针对键值对的数据集进行使用

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 2), ("kitty", 3)))
    val rdd2 = sc.parallelize(List(("jerry", 9), ("tom", 8), ("shuke", 7)))
    
    val rdd3 = rdd1.join(rdd2)
    val rdd3 = rdd1.leftOuterJoin(rdd2)
    val rdd3 = rdd1.rightOuterJoin(rdd2)

--------------------

### groupByKey ###

val rdd3 = rdd1 union rdd2
    rdd3.groupByKey
    rdd3.groupByKey.map(x=>(x._1,x._2.sum))
    
    sc.textFile("hdfs://bigdata01:9000/spark").flatMap(x=>x.split("")).map((_,1)).reduceByKey(_+_).sortBy(_._2,false).collect
    sc.textFile("hdfs://bigdata01:9000/spark").flatMap(x=>x.split("")).map((_,1)).groupByKey.map(t=>(t._1, t._2.sum)).collect

--------------------

### cogroup ###

val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
    val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
    val rdd3 = rdd1.cogroup(rdd2)
    val rdd4 = rdd3.map(t=>(t._1, t._2._1.sum + t._2._2.sum))

--------------------

### cartesian笛卡尔积 ###

val rdd1 = sc.parallelize(List("tom", "jerry"))
    val rdd2 = sc.parallelize(List("tom", "kitty", "shuke"))
    val rdd3 = rdd1.cartesian(rdd2)

--------------------

### filter ###

def filter(f: (T) ⇒ Boolean): RDD\[T\]

--------------------

### map ###

def map\[U\](f: (T) ⇒ U)(implicit arg0: ClassTag\[U\]): RDD\[U\]  
传进来一个函数，指示对数据集中每一个元素进行的操作

val arr1 = Array(1,2,3,4,5)
    val arr2 = rdd1.map(_+1)

结果会输出Array(2,3,4,5,6)  
也就是对arr1类里面的每一个元素都加上了1

--------------------

### mapValues ###

只对值进行操作，在对键值对数据集使用的时候特别方便

--------------------

### mapPartitions ###

> def  
> mapPartitions\[U\](f: (Iterator\[T\]) ⇒ Iterator\[U\], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag\[U\]): RDD\[U\]

map是对每个元素操作, mapPartitions是对其中的每个partition分别操作  
对rdd和数据集作用

参数是Iterator类型，可以通过一个返回值为Iterator的函数，或者转化成List然后用List 的.iterator方法，通常是写函数然后传进入，就好像示例1那样

--------------------

示例1：

val a = sc.parallelize(1 to 9, 3)
      def doubleFunc(iter: Iterator[Int]) : Iterator[(Int,Int)] = {
        var res = List[(Int,Int)]()
        while (iter.hasNext)
        {
          val cur = iter.next;
          res .::= (cur,cur*2)
        }
        res.iterator
      }
    val result = a.mapPartitions(doubleFunc)//这里输入的是上面定义的函数，在这里不用给这个函数传参数
    println(result.collect().mkString)

示例2

val rdd1= sc.parallelize(  1 to 9,2)
    val rdd2  = rdd1.mapPartition(x => List(x).iterator)
    rdd2.collect

--------------------

### mapPartitionsWithIndex ###

> def  
> mapPartitionsWithIndex\[U\](f: (Int, Iterator\[T\]) ⇒ Iterator\[U\], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag\[U\]): RDD\[U\]

与mapPartitions相似，只不过可以根据分区的索引进行处理  
对rdd和数据集进行操作，返回跟原操作对象相同的对象

参数是Iterator类型，可以通过一个返回值为Iterator的函数，或者转化成List然后用List 的.iterator方法，通常是写函数然后传进入，就好像示例1那样

示例1

var rdd1 = sc.makeRDD(1 to 5,2)
        //rdd1有两个分区
        var rdd2 = rdd1.mapPartitionsWithIndex{
                (x,iter) => {
                  var result = List[String]()
                    var i = 0
                    while(iter.hasNext){
                      i += iter.next()
                    }
                    result.::(x + "|" + i).iterator
    
                }
              }
        //rdd2将rdd1中每个分区的数字累加，并在每个分区的累加结果前面加了分区索引
        scala> rdd2.collect
        res13: Array[String] = Array(0|3, 1|12)

示例2

def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {
      iter.toList.map(x =>"[partID:" +  index + ", val: " + x + "]").iterator
    }
    val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
    rdd1.mapPartitionsWithIndex(func1).collect

--------------------

### aggregate ###

聚合

#### 是action操作, 第一个参数是初始值zerovalue, 第二个参数是2个函数 ，每个函数都是2个参数 。第一个函数先对各个分区进行合并, 第二个函数对各个分区合并后的结果再进行合并), ####

> val rdd1= sc.parallelize( 1 to 9,2)
> 
> rdd1.aggregate(0)(*+*, *+*)

这个语句执行的时候，其实是这样的 0 + (0+(1+2+3+4)) + (0+(5+6+7+8+9))  
第一个参数zerovalue在这里为0，也就是每次执行函数的时候，首先将每个分区里面的所有数字加起来：(1+2+3+4)，此时每个分区结果还要和zerovalue 0 来进行运算，变成 0+（1+2+3+4），0+（5+6+7+8+9），然后执行第二个函数，将分区结果再相加：  
0+（1+2+3+4）+ 0+（5+6+7+8+9），此时还要和zerovalue 进行运算，变成0+ (0+（1+2+3+4）+ 0+（5+6+7+8+9）)  
执行完后输出的结果是Int=45

> rdd1.aggregate(1)(\_-\_,\_+\_)

这条语句执行完后输出的结果是-42，这个和上一个一对比就能明显的知道执行的过程了，先对每一个分区里的所有元素进行运算(-1-2-3-4),(-5-6-7-8-9),然后每个分区要和zerovalue 1 进行运算，这里用的运算函数是第二个函数，因为已经得到每个分区结果了，使用*+*对分区结果进行合并，也就是1+(-1-2-3-4)+1+(-5-6-7-8-9),最后和zerovalue 进行运算 1+1+(-1-2-3-4)+1+(-5-6-7-8-9) 也就是-42

总结：对分区内每个元素进行运算，用第一个函数，然后和zerovalue进行运算（用第二个函数），然后对分区结果进行合并，用第二个函数。

> rdd1.aggregate(0)(math.max(\_, \_), \_ + \_)  
> 这句就比较好理解了，求出每个分区最大的元素，然后把每个分区最大的元素相加，输出结果是13 (4+9)
> 
> rdd1.aggregate(5)(math.max(\_, \_), \_ + \_)  
> 5和1比, 得5再和234比得5 –> 5和6789比,得9 –> 5 + (5+9)

下面是其他一些例子

val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
    
    rdd2.aggregate("")(_ + _, _ + _)
    rdd2.aggregate("=")(_ + _, _ + _)
    
    val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
    rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y) val rdd4 = sc.parallelize(List("12","23","345",""),2) rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y) val rdd5 = sc.parallelize(List("12","23","","345"),2) rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)

--------------------

### aggregateByKey ###

针对键值对的列表使用，根据键来进行聚合，先进行聚合，然后进行函数操作

> val pairRDD = sc.parallelize(List( (“cat”,2), (“cat”, 5), (“mouse”, 4),(“cat”, 12), (“dog”, 12), (“mouse”, 2)), 2)  
> pairRDD.aggregateByKey(0)(math.max(\_, \_), \_ + \_).collect

在这里会根据每个分区内的每个键值对的键进行聚合，变成 List((“cat”,19),(“mouse”,6),(“dog”,12))

--------------------

### repartition(numPartitions:Int):RDD\[T\] coalesce(numPartitions:Int，shuffle:Boolean=false):RDD\[T\] ###

对RDD的分区进行重新划分，repartition只是coalesce接口中shuffle为true的简易实现

val rdd1 = sc.parallelize(1 to 10, 10)
    val rdd2 = rdd1.coalesce(2, false)
    rdd2.partitions.length

如果新分区的数量与原分区相差不大，就不需要shuffle，也就是把shuffle设为false

--------------------

### collectAsMap : Map(b -> 2, a -> 1) ###

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
    rdd.collectAsMap

--------------------

### combineByKey ###

def combineByKey\[C\](  
createCombiner: V => C,  
mergeValue: (C, V) => C,  
mergeCombiners: (C, C) => C,  
partitioner: Partitioner,  
mapSideCombine: Boolean = true,  
serializer: Serializer = null)  
一共有6个参数。如下解释下3个重要的函数参数：  
createCombiner: V => C ，这个函数把当前的值作为参数，此时我们可以对其做些附加操作(类型转换)并把它返回 (这一步类似于初始化操作)  
mergeValue: (C, V) => C，该函数把元素V(下一个将要处理的值)合并到之前的元素C(createCombiner)上 (这个操作在每个分区内进行，对分区内部的元素进行操作)  
mergeCombiners: (C, C) => C，该函数把2个元素C(两个分区的已经合并的元素)合并 (这个操作在不同分区间进行)

每个分区中每个key中value中的第一个值, (hello,1)(hello,1)(good,1) –> (hello(1,1),good(1)) –> x就相当于hello的第一个1, good中的1

示例1：

> val rdd1 = sc.textFile(“hdfs://bigdata01:9000/spark/”).flatMap(\_.split(“”)).map((\_, 1))  
> val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)  
> rdd1.collect  
> rdd2.collect

示例2

> val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)  
> rdd3.collect

val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
    val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
    val rdd6 = rdd5.zip(rdd4)
    val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)

--------------------

### countByKey ###

统计数据集中每个Key有多少个键值对  
针对键值对的数据集

--------------------

### filterByRange ###

获取数据集中key为某一个范围内的元素  
作用于rdd 传入key的开始和结束值

> val rdd1 = sc.parallelize(List((“e”, 5), (“c”, 3), (“d”, 4), (“c”, 2), (“a”, 1)))  
> val rdd2 = rdd1.filterByRange(“b”, “d”)  
> rdd2.collect

--------------------

### flatMapValues ###

将键值对中的值扁平化

> val rdd3 = sc.parallelize(List((“a”, “1 2”), (“b”, “3 4”),(“a”, “1 6”)))  
> val rdd4 = rdd3.flatMapValues(\_.split(“”))  
> rdd4.collect

输出结果为：Array\[(String, String)\] = Array((a,1), (a,” “), (a,2), (b,3), (b,” “), (b,4), (a,1), (a,” “), (a,6))

--------------------

### foldByKey(初始值)(操作函数) ###

将键值对根据相同的键值进行折叠  
第一个参数是一个zerovalue，定义对键的初始操作，第二个参数为函数，可以定义如何进行折叠

> val rdd1 = sc.parallelize(List(“dog”, “wolf”, “cat”, “bear”), 2)  
> val rdd2 = rdd1.map(x => (x.length, x))  
> val rdd3 = rdd2.foldByKey(” “)(\_+\_).collect  
> 输出为 Array((3,2),(4,2))这样的数组

--------------------

### foreachPartition(操作函数) ###

可以对每个分区单独操作，比如将一个分区里面的元素全部累加等

> val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)  
> rdd1.foreachPartition(x => println(x.reduce(\_ + \_)))  
> 输出为类似 6 12 24 这样的

--------------------

keyBy : 以传入的参数做key

val rdd1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
    val rdd2 = rdd1.keyBy(_.length)
    rdd2.collect

--------------------

keys values
    val rdd1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
    val rdd2 = rdd1.map(x => (x.length, x))
    rdd2.keys.collect
    rdd2.values.collect

--------------------

### zip ###

def zip\[U\](other: RDD\[U\])(implicit arg0: ClassTag\[U\]): RDD\[(T, U)\]  
zip函数用于将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常

rdd1.zip(rdd2) 则rdd1的元素将会作为Key，rdd2的元素将会作为value

scala> var rdd1 = sc.makeRDD(1 to 10,2)  
    rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at :21  
    
    scala> var rdd1 = sc.makeRDD(1 to 5,2)  
    rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at makeRDD at :21  
    
    scala> var rdd2 = sc.makeRDD(Seq("A","B","C","D","E"),2)  
    rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[2] at makeRDD at :21  
    
    scala> rdd1.zip(rdd2).collect  
    res0: Array[(Int, String)] = Array((1,A), (2,B), (3,C), (4,D), (5,E))             
    
    scala> rdd2.zip(rdd1).collect  
    res1: Array[(String, Int)] = Array((A,1), (B,2), (C,3), (D,4), (E,5))  
    
    scala> var rdd3 = sc.makeRDD(Seq("A","B","C","D","E"),3)  
    rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[5] at makeRDD at :21  
    
    scala> rdd1.zip(rdd3).collect  
    java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions  
    //如果两个RDD分区数不同，则抛出异常