SparkCore（三）：RDD转换算子

悠悠 2022-12-10 12:25 222阅读 0赞

### 文章目录 ###

*  一、Value类型
 *   *  1.1 map
     *  1.2 mapPartitions
     *  1.3 mapPartitionsWithIndex
     *  1.4 flatMap
     *  1.5 glom
     *  1.6 groupBy
     *  1.7 filter
     *  1.8 sample
     *  1.9 distinct
     *  1.10 coalesce
     *  1.11 repartition
     *  1.12 sortBy
 *  二、双Value类型
 *   *  2.1 intersection
     *  2.2 union
     *  2.3 subtract
     *  2.4 zip
 *  三、Key-Value类型
 *   *  3.1 partitionBy
     *  3.2 reduceByKey
     *  3.3 groupByKey
     *  3.4 aggregateByKey
     *  3.5 foldByKey
     *  3.6 combineByKey
     *  3.7 sortByKey
     *  3.8 join
     *  3.9 leftOuterJoin
     *  3.10 cogroup
 *  四、案例实操

> RDD根据数据处理方式的不同将算子整体上分为Value类型、双Value类型和Key-Value类型。

# 一、Value类型 #

## 1.1 map ##

函数签名：`def map[U: ClassTag](f: T => U): RDD[U]`

函数说明：将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换

val sparkConf = new SparkConf().setMaster("local").setAppName("RDDTest")
        val sc = new SparkConantext(sparkConf)
        val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
        val rdd2 = rdd1.map(x => { 
          println("first ==> " + x)
          x
        })
        rdd2.map(x => { 
          println("second ==>" + x)
          x
        }).collect()
        sc.stop()

输出：

A ==> 2
    A ==> 1
    B ==>2
    B ==>1
    A ==> 3
    B ==>3

多次测试，我们可以得出结论：分区内数据是按照顺序依次执行，第一条数据所有逻辑全部执行完毕后才会执行下一条数据；分区间数据执行没有顺序，而且无需等待。

## 1.2 mapPartitions ##

**函数签名：**

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

**函数说明**：将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
        rdd.mapPartitions(iter => { 
          iter.foreach(println)
          iter
        }).collect()

## 1.3 mapPartitionsWithIndex ##

**函数签名：**

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

**函数说明：** 将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据，在处理时同时可以获取当前分区索引

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3), 2)
        rdd.mapPartitionsWithIndex((index, iter) => { 
          println("当前分区："+index)
          iter.foreach(t => println(index + "->" + t))
          iter
        }).collect()

打印：

当前分区：0
    0->1
    当前分区：1
    1->2
    1->3

## 1.4 flatMap ##

**函数签名**：`def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]`

**函数说明**：将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射

/** 统计字母出现的次数第一步：扁平化操作 */
        val rdd: RDD[String] = sc.makeRDD(List("1 a", "2 b", "3 c"), 2)
        rdd.flatMap(t => { 
          val char = t.split(" ")(1)
          char.map((_, 1))
        }).collect().foreach(println)

打印：

(a,1)
    (b,1)
    (c,1)

## 1.5 glom ##

**函数签名**：`def glom(): RDD[Array[T]]`

**函数说明**：将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变

/** 求分区内最大值 */
        val rdd: RDD[Int] = sc.parallelize(List(1, 2, 3, 5, 4, 6), 3)
        rdd.glom().collect().foreach(iter => println("分区内最大值" + iter.max))

打印：

分区内最大值：2
    分区内最大值：5
    分区内最大值：6

## 1.6 groupBy ##

**函数签名**：`def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]`

**函数说明**：将数据根据指定的规则进行分组，分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为`shuffle`。极限情况下，数据可能被分在同一个分区中。一个组的数据在一个分区中，但是并不是说一个分区中只有一个组

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 5, 4, 6), 3)
        rdd.groupBy(t => t % 2).collect().foreach(println)

打印：

(0,CompactBuffer(2, 4, 6))
    (1,CompactBuffer(1, 3, 5))

## 1.7 filter ##

**函数签名**：`def filter(f: T => Boolean): RDD[T]`

**函数说明**：将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

val rdd: RDD[Int] = sc.makeRDD(List(0, 0, 0, 1, 1, 1), 2)
        rdd.filter(t => t == 0).saveAsTextFile("output")

查看保存文件：此时已发生数据倾斜  
![在这里插入图片描述][20200928164447753.png]

## 1.8 sample ##

**函数签名**：

def sample(
      withReplacement: Boolean,
      fraction: Double,
      seed: Long = Utils.random.nextLong): RDD[T]

**参数说明：**

*  第一个参数：抽取的数据是否放回
 *  第二个参数：抽取的几率，范围在\[0，1\]之间，0：全不取；1：全取；
 *  第三个参数：随机数种子，添加随机数种子生产的随机数都是伪随机，都是固定值

**函数说明**：根据指定的规则从数据集中抽取数据

**作用**：在实际开发中，往往会出现数据倾斜的情况，那么可以从数据倾斜的分区中抽取数据，查看数据的规则，分析后，可以进行改善处理，让数据更加均匀

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5), 1)
        /** * 伯努利算法（抽取数据不放回） */
        rdd.sample(false, 0.5).collect().foreach(t => print(t + " "))
        println()
        /** * 泊松算法（抽取数据放回） */
        rdd.sample(true, 0.5).collect().foreach(t => print(t + " "))

打印：

3 5 
    1 2 2 3

## 1.9 distinct ##

**函数签名**：

def distinct()(implicit ord: Ordering[T] = null): RDD[T]
    def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

**函数说明**：将数据集中重复的数据去重

val rdd: RDD[Int] = sc.makeRDD(List(0, 0, 0, 1, 1, 1), 1)
        rdd.distinct().saveAsTextFile("output")
        //将去重后数据进行分区
        rdd.distinct(2).saveAsTextFile("output2")

## 1.10 coalesce ##

**函数签名**：

def coalesce(numPartitions: Int, shuffle: Boolean = false,
               partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
              (implicit ord: Ordering[T] = null)
      : RDD[T]

**参数说明：**

*  第一个参数：分区数
 *  第二个参数：是否`shuffle`

**函数说明**：根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率。

**作用**：当`spark`程序中，存在过多的小任务的时候，可以通过`coalesce`方法，收缩合并分区，减少分区的个数，减小任务调度成本

val rdd: RDD[Int] = sc.makeRDD(List(0, 0, 0, 1, 1, 1), 3)
        rdd.filter(t => t == 0).saveAsTextFile("output")
        rdd.filter(t => t == 0).coalesce(2,true).saveAsTextFile("output2")

## 1.11 repartition ##

**函数签名**：`def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]`

**函数说明**：该操作内部其实执行的是`coalesce`操作，参数`shuffle`的默认值为`true`。无论是将分区数多的`RDD`转换为分区数少的`RDD`，还是将分区数少的`RDD`转换为分区数多的`RDD`，`repartition`操作都可以完成，因为无论如何都会经`shuffle`过程。

**coalesce和repartition对比**：`repartition`方法其实就是`coalesce`方法，只不过肯定使用了`shuffle`操作。让数据更均衡一些，可以有效防止数据倾斜问题。如果缩减分区，一般就采用`coalesce`，如果扩大分区，就采用`repartition`

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5), 2)
        rdd.repartition(3).saveAsTextFile("output")

## 1.12 sortBy ##

**函数签名**：

def sortBy[K](
      f: (T) => K,
      ascending: Boolean = true,
      numPartitions: Int = this.partitions.length)
      (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

**参数说明：**

*  第一个参数：排序函数
 *  第二个参数：是否正序排序
 *  第三个参数：分数数

**函数说明**：该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为正序排列。排序后新产生的`RDD`的分区数与原`RDD`的分区数一致。

val rdd: RDD[Int] = sc.makeRDD(List(5, 3, 4, 1, 2), 2)
        rdd.sortBy(t => t, false, 3).saveAsTextFile("output")

# 二、双Value类型 #

## 2.1 intersection ##

**函数签名**：`def intersection(other: RDD[T]): RDD[T]`

**函数说明**：对源`RDD`和参数`RDD`求交集后返回一个新的`RDD`

val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
        val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
    
        val newRdd = rdd1.intersection(rdd2)
        newRdd.collect().foreach(println)

打印：

4
    3

## 2.2 union ##

**函数签名**：`def union(other: RDD[T]): RDD[T]`

**函数说明**：对源`RDD`和参数`RDD`求并集后返回一个新的`RDD`

val sc = new SparkContext(sparkConf)
        val rdd1 = sc.makeRDD(List(1, 2))
        val rdd2 = sc.makeRDD(List(2, 3))
    
        val newRdd = rdd1.union(rdd2)
        newRdd.collect().foreach(println)

打印：

1
    2
    2
    3

## 2.3 subtract ##

**函数签名**：`def subtract(other: RDD[T]): RDD[T]`

**函数说明**：以一个`RDD`元素为主，去除两个`RDD`中重复元素，将其他元素保留下来。求差集

val rdd1 = sc.makeRDD(List(1, 2, 3, 4))
        val rdd2 = sc.makeRDD(List(3, 4, 5, 6))
    
        val newRdd = rdd1.subtract(rdd2)
        newRdd.collect().foreach(println)

打印：

1
    2

## 2.4 zip ##

**函数签名**：`def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]`

**函数说明**：将两个`RDD`中的元素，以键值对的形式进行合并。其中，键值对中的`Key`为第1个`RDD`中的元素，`Value`为第2个`RDD`中的元素。

val rdd1 = sc.makeRDD(List(1, 2, 3))
        val rdd2 = sc.makeRDD(List(3, 4, 5))
    
        val newRdd = rdd1.zip(rdd2)
        newRdd.collect().foreach(println)

打印：

(1,3)
    (2,4)
    (3,5)

如果两个`RDD`数据类型不一致怎么办？ 正常

如果两个`RDD`数据分区不一致怎么办？报错`can't zip RDDs with unequal numbers of partitions`

如果两个`RDD`分区数据数量不一致怎么办？ 报错`Can only zip RDDs with same number of elements in each partition`

# 三、Key-Value类型 #

## 3.1 partitionBy ##

**函数签名**：`def partitionBy(partitioner: Partitioner): RDD[(K, V)]`

**函数说明**：将数据按照指定`Partitioner`重新进行分区。`Spark`默认的分区器是`HashPartitioner`

val rdd = sc.makeRDD(List((2, "b"), (1, "a"), (3, "c")), 1)
        val newRdd = rdd.partitionBy(new HashPartitioner(2))
        newRdd.saveAsTextFile("output")

如果重分区的分区器和当前`RDD`的分区器一样怎么办？ 不进行任何处理，不会再次重分区

`Spark`还有其他分区器吗？ `SortBy`默认使用`RangePartitioner`

如果想按照自己的方法进行数据分区怎么办？自定义分区器，如下：

class MyPartitioner(num: Int) extends Partitioner { 
        override def numPartitions: Int = num
    
        override def getPartition(key: Any): Int = { 
          key match { 
            case "1" => 0
            case _ => 1
          }
        }
      }

## 3.2 reduceByKey ##

**函数签名**：

def reduceByKey(func: (V, V) => V): RDD[(K, V)]
    def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

**函数说明**： 可以将数据按照相同的`Key`对`Value`进行聚合

val rdd = sc.makeRDD(List(("a", 1), ("b", 1), ("a", 1)), 1)
        rdd.reduceByKey(_ + _).foreach(println)
        rdd.reduceByKey(_ + _,2).foreach(println)

打印：

(a,2)
    (b,1)
    (a,2)
    (b,1)

## 3.3 groupByKey ##

**函数签名**：

def groupByKey(): RDD[(K, Iterable[V])]
    def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
    def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]

**函数说明**： 将分区的数据直接转换为相同类型的内存数组进行后续处理

val rdd = sc.makeRDD(List(("a", 1), ("b", 1), ("a", 1)), 1)
        val newRdd: RDD[(String, Iterable[Int])] = rdd.groupByKey()
        newRdd.map(t => (t._1, t._2.size)).foreach(println)

打印：

(a,2)
    (b,1)

## 3.4 aggregateByKey ##

**函数签名**：

def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
      combOp: (U, U) => U): RDD[(K, U)]

![在这里插入图片描述][20200929200520581.png]

**函数说明**： 将数据根据不同的规则进行分区内计算和分区间计算

/** * 取出每个分区内相同 key 的最大值然后分区间相加 */
       val rdd = sc.makeRDD(List(("a", 1), ("a", 2), ("c", 3),
          ("b", 1), ("b", 2), ("c", 3)
        ), 2)
        val newRdd = rdd.aggregateByKey(0)(
          (x, y) => math.max(x, y),
          (x, y) => x + y)
        println(newRdd.collect().mkString(","))

## 3.5 foldByKey ##

**函数签名**：`def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]`

**函数说明：** 当分区内计算规则和分区间计算规则相同时，`aggregateByKey`就可以简化为`foldByKey`

val rdd = sc.makeRDD(List(("a", 1), ("a", 2), ("c", 3),
          ("b", 1), ("b", 2), ("c", 4)
        ), 2)
        /** * (a,2),(c,3) } (a,2) * (b,2),(c,4) } (b,2)(c,4) */
        val newRdd = rdd.foldByKey(0)(
          (x, y) => math.max(x, y))
        println(newRdd.collect().mkString(","))

打印：

(b,2),(a,2),(c,4)

## 3.6 combineByKey ##

**函数签名：**

def combineByKey[C](
      createCombiner: V => C,
      mergeValue: (C, V) => C,
      mergeCombiners: (C, C) => C): RDD[(K, C)]

**参数说明：**

*  第一个参数：将计算的第一个值转换结构
 *  第二个参数：分区内计算规则
 *  第三个参数：分区间计算规则

**函数说明**：最通用的对`key-value`型`RDD`进行聚集操作的聚集函数（`aggregation function`）。类似于`aggregate()`，`combineByKey()`允许用户返回值的类型与输入不一致。

/** 求每个key的平均值 */
        val rdd = sc.makeRDD(List(("a", 88), ("b", 95), ("a", 91),
          ("b", 93), ("a", 95), ("b", 98)), 2)
    
        val newRdd = rdd.combineByKey((_, 1),
          (temp: (Int, Int), v) => (temp._1 + v, temp._2 + 1),
          (t1: (Int, Int), t2: (Int, Int)) => (t1._1 + t2._1, t1._2 + t2._2)
        )
        val resultRdd = newRdd.map(t => (t._1, t._2._1 / t._2._2))
        println(resultRdd.collect().mkString(","))

打印：

(b,95),(a,91)

## 3.7 sortByKey ##

**函数签名：** `def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length) : RDD[(K, V)]`

**函数说明**：在一个`(K,V)`的`RDD`上调用，`K`必须实现`Ordered`接口，返回一个按照`key`进行排序的

val rdd = sc.makeRDD(List(("a", 1), ("c", 2), ("b", 3)))
        val newRdd1 = rdd.sortByKey(true)
        val newRdd2 = rdd.sortByKey(false)
        println(newRdd1.collect().mkString(","))
        println(newRdd2.collect().mkString(","))

打印：

(a,1),(b,3),(c,2)
    (c,2),(b,3),(a,1)

设置`key`为自定义类：

class Stu extends Ordered[Stu]{ 
      override def compare(that: Stu): Int = { 
          1
        }
    }

## 3.8 join ##

**函数签名：** `def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]`

**函数说明**：在类型为`(K,V)`和`(K,W)`的`RDD`上调用，返回一个相同`key`对应的所有元素连接在一起的`(K,(V,W))`的`RDD`

val rdd7 = sc.makeRDD(List(("a", "a"), ("b", "b"),("c", "c")))
        val rdd8 = sc.makeRDD(List(("a", 1), ("b", 2)))
        rdd7.join(rdd8).collect().foreach(println)

打印：

(a,(a,1))
    (b,(b,2))

如果`key`存在不相等呢？  
如果`key`不相等，对应的数据无法连接，如果`key`有重复的，那么数据会多次连接

## 3.9 leftOuterJoin ##

**函数签名：** `def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]`

**函数说明**：类似于`SQL`语句的左外连接

val rdd1 = sc.makeRDD(List(("a", 1), ("b", 2),("c", 3)))
        val rdd2 = sc.makeRDD(List(("a", 1), ("b", 2)))
        val newRdd = rdd1.leftOuterJoin(rdd2)
        newRdd.collect().foreach(println)

打印：

(a,(1,Some(1)))
    (b,(2,Some(2)))
    (c,(3,None))

## 3.10 cogroup ##

**函数签名：** `def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]`

**函数说明**：在类型为`(K,V)`和`(K,W)`的`RDD`上调用，返回一个`(K,(Iterable<V>,Iterable<W>))`类型的`RDD`

val rdd1 = sc.makeRDD(List(("a", 1), ("b", 2),("c", 3),("c", 4)))
        val rdd2 = sc.makeRDD(List(("a", 1), ("b", 2), ("b", 3)))
        val newRdd = rdd1.cogroup(rdd2)
        newRdd.collect().foreach(println)

打印：

(a,(CompactBuffer(1),CompactBuffer(1)))
    (b,(CompactBuffer(2),CompactBuffer(2, 3)))
    (c,(CompactBuffer(3, 4),CompactBuffer()))

# 四、案例实操 #

**需求描述：** 统计出每一个省份每个广告被点击数量排行的Top3

**需求分析：**  
![在这里插入图片描述][20201010173145310.png]  
**功能实现：**

val sparkConf = new SparkConf().setMaster("local").setAppName("RDDDemo")
        val sc = new SparkContext(sparkConf)
        //1.读取数据
        val lines: RDD[String] = sc.textFile("input/agent.log")
        //2.将数据转换成我们想要的数据
        val effectData = lines.map(line => { 
          val words = line.split(" ")
          ((words(1), words(4)), 1)
        })
        //3.reduceByKey累加
        val calculateData = effectData.reduceByKey(_ + _)
        //4.对数据进行映射
        val newData = calculateData.map(t => (t._1._1, (t._1._2, t._2)))
        //5.根据省份进行分组
        val testData = newData.groupByKey()
        //6.排序并取前三位数据
        val resultData = testData.mapValues(iter => { 
          iter.toList.sortWith((left, right) => { 
            left._2 > right._2
          }).take(3)
        })
        resultData.collect().foreach(println)

[20200928164447753.png]: /images/20221123/34ad2d5369f0477ba7f90e4eb72bc2e7.png
[20200929200520581.png]: /images/20221123/0eb9d04a3f234821a028a03930922486.png
[20201010173145310.png]: /images/20221123/fce857a12b0e4c3794ca12f27d87fbc6.png