五-, Spark 算子 (转化+行动算子共三十七个)

五-中, Spark 算子吐血总结

5.1.4.3 RDD 转换算子(Transformation)

什么是算子?

在流处理、交互式查询中有个常用的概念是“算子”，在英文中被成为“Operation”，在数学上可以解释为一个函数空间到另一个函数空间上的映射O：X->X，其实就是一个处理单元，往往是指一个函数，在使用算子时往往会有输入和输出，算子则完成相应数据的转化，比如：Group、Sort等都是算子。

---

从大方向来说, Spark算子(RDD方法)大致可以分为以下两类:

1. Transformation 变换/转换算子 : 这种变换并不触发提交作业, 而是完成作业中间过程处理; Transformation 操作是延迟计算的, 也就是说从一个RDD转换为另一个RDD的转换操作不是马上执行, 需要等到有Action操作(行动算子)的时候才会真正触发运算;
2. Action 行动算子: 这类算子会触发SparkContext 提交Job作业, 并将数据输出到Spark系统;

从小方向来说, Spark算子大致分为三类:

1. Value数据类型的Transformation算子, 这种变换并不触发提交作业, 针对处理的数据项是Value型的数据;
2. Key-Value数据类型的Transformation算子, 这种变换并不触发提交作业, 针对处理的数据项是Key-Value型的数据对;
3. Action算子, 这类算子会触发SparkContext提交Job作业;

1. Value类型

1.1 map

val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD1: RDD[Int] = dataRDD.map(num => {
  num * 2
})
val dataRDD2: RDD[String] = dataRDD1.map(num => {
  ""   num
})

map的并行计算🌰:
// 1. rdd的计算一个分区内的数据是一个一个执行逻辑
// 只有前面一个数据全部的逻辑执行完毕后，才会执行下一个数据。
// 分区内数据的执行是有序的。
// 2. 不同分区数据计算是无序的。

1.2 mapPartitions

栗子: 获取每个数据分区的最大值

def main(args: Array[String]): Unit = {
    //1. 新建配置文件对象
    val conf: SparkConf = new SparkConf()

    conf.setMaster("local[*]")
    conf.setAppName("memoryRDD")
    //2. 新建sparkContext
    val sc: SparkContext = new SparkContext(conf)

    //3. 读取文件
    val rdd= sc.makeRDD(List(1,2,3,4,5,6,7,8,9), 5)

    //由上面讲过的内存中数据的分区规则
    /**分区号: 数据
     * 0: 1
     * 1: 2,3
     * 2: 4,5
     * 3: 6,7
     * 4: 8,9
     */
    //mapPartitons()的入参和出参都是迭代器!!!!!
    //mapOartitions()一次计算的是一个分区的数据
    val mapRDD:RDD[Int] = rdd.mapPartitions(
      //iter.max 求出每个分区的最大值, 是int
      //但是这个函数要求出参是迭代器!
      iter => List(iter.max).iterator
    )
      // mapPartitions : 可以以分区为单位进行数据转换操作
        //                 但是会将整个分区的数据加载到内存进行引用
        //                 如果处理完的数据是不会被释放掉，存在对象的引用。
        //                 在内存较小，数据量较大的场合下，容易出现内存溢出。
    //4. collect 从内存中收集数据, 迭代输出
    mapRDD.collect().foreach(println)

    mapRDD.saveAsTextFile("output")
    //5. 关闭资源
    sc.stop()
}

Q: map 和 mapPartitions的区别 ?

1.3 mapPartitionsWithIndex

举个🌰:

//就记住这一点就会用了
// mapWithPartitionsIndex, 入参是(index, iterator), 出参是 iterator
def main(args: Array[String]): Unit = {
    //1. 新建配置文件对象
    val conf: SparkConf = new SparkConf()

    conf.setMaster("local[*]")
    conf.setAppName("memoryRDD")
    //2. 新建sparkContext
    val sc: SparkContext = new SparkContext(conf)

    //3. 读取文件
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4), 2)

    val rddWithIndex: RDD[Int] = rdd.mapPartitionsWithIndex(
      (index, iterator) => {
        index match {
          case 1 => iterator
          case _ => Nil.iterator
        }
      }
    )

    //4. 存储结果到文件
    rddWithIndex.collect().foreach(println)
    rddWithIndex.saveAsTextFile("output")

    //5. 关闭资源
    sc.stop()
  }

如何直接打印数据所在的分区?

val mpiRDD = rdd.mapPartitionsWithIndex(
            (index, iter) => {
                // 1,   2,    3,   4
                //(0,1)(2,2),(4,3),(6,4)
                iter.map(
                    num => {
                        (index, num)
                    }
                )
            }
        )

1.4 flatMap

def main(args: Array[String]): Unit = {
    //1. 新建配置文件对象
    val conf: SparkConf = new SparkConf()

    conf.setMaster("local[*]")
    conf.setAppName("memoryRDD")
    //2. 新建sparkContext
    val sc: SparkContext = new SparkContext(conf)

    //3. 读取文件
    val rdd = sc.makeRDD(List(List(1,2),3,List(4,5)))

    val rddMap = rdd.flatMap(dat => {
      dat match {
        case i: Int => List(i)
        case j: List[_] => j
      }
    })

    //4. 存储结果到文件
    rddMap.saveAsTextFile("output")
    rddMap.collect().foreach(println)
    //5. 关闭资源
    sc.stop()
}

1.5 glom

计算所有分区最大值求和（分区内取最大值，分区间最大值求和）

def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
        val sc = new SparkContext(sparkConf)

        // TODO 算子 - glom
        val rdd : RDD[Int] = sc.makeRDD(List(1,2,3,4), 2)

        // 【1，2】，【3，4】
        // 【2】，【4】
        // 【6】
        val glomRDD: RDD[Array[Int]] = rdd.glom()

        val maxRDD: RDD[Int] = glomRDD.map(
            array => {
                array.max
            }
        )
        println(maxRDD.collect().sum)
        sc.stop()
    }
}

1.6 groupBy

一个组的数据在一个分区中, 但是并不是说一个分区中只有一个组;

1. List(“Hello”, “hive”, “hbase”, “Hadoop”)根据单词首写字母进行分组。

sc.groupBy(_.charAt(0))

2. 从服务器日志数据 apache.log 中获取每个时间段访问量。

//1. 配置文件对象
    val conf = new SparkConf();
    conf.setAppName("groupby")
    conf.setMaster("local[*]")
    //2. sparkcontext
    val sc = new SparkContext(conf)

    //3. 创建RDD
    val rddString: RDD[String] = sc.textFile("datas/apache.log")

    //4.
    val hourRDD: RDD[(String, Int)] = rddString.map(
      line => {
        //取出时间
        val strArr = line.split(" ")
        val time = strArr(3)

        //对time格式化
        //指定格式化模式
        val sdfOfDate = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
        //格式化时间
        val date: Date = sdfOfDate.parse(time)

        //指定格式化模式
        val sdfOfHour = new SimpleDateFormat("HH")
        //格式化时间, 获取到了小时
        val hour: String = sdfOfHour.format(date)
        (hour, 1)
      }
    )

    //5. 分组
    val hourOfGroupedRDD: RDD[(String, Iterable[(String, Int)])] = hourRDD.groupBy(_._1)

    //6. 汇总计算
    val res: RDD[(String, Int)] = hourOfGroupedRDD.map {
      case (hour: String, iter: Iterable[(String, Int)]) =>
        (hour, iter.size)
    }
    res.collect().foreach(println)

3. WordCount

1.7 filter

简单例子:

val dataRDD = sparkContext.makeRDD(List(1,2,3,4),1)
val dataRDD1 = dataRDD.filter(_%2 == 0)

从服务器日志数据 apache.log 中获取 2015 年 5 月 17 日的请求路径

//1. 配置文件对象
    val conf = new SparkConf();
    conf.setAppName("groupby")
    conf.setMaster("local[*]")
    //2. sparkcontext
    val sc = new SparkContext(conf)

    //3. 创建RDD
    val rddString: RDD[String] = sc.textFile("datas/apache.log")

    //4.
    val timeAndPathRDD: RDD[(String, String)] = rddString.map(
      line => {
        //取出时间
        val strArr = line.split(" ")
        val timeAndPathTuple: Tuple2[String, String] = (strArr(3), strArr(6))

        //对time格式化
        //指定格式化模式
        val sdfOfDate = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
        //格式化时间
        val date: Date = sdfOfDate.parse(timeAndPathTuple._1)

        //指定格式化模式
        val sdfOfTime = new SimpleDateFormat("dd/MM/yyyy")
        //格式化时间, 获取到了小时
        val time: String = sdfOfTime.format(date)
        (time, timeAndPathTuple._2)
      }
    )
    //过滤
    val res = timeAndPathRDD.filter(_._1.equals("17/05/2015"))
    res.collect().foreach(println)

1.8 sample

val dataRDD = sparkContext.makeRDD(List(1,2,3,4),1)
    // 抽取数据不放回（伯努利算法）
    // 伯努利算法：又叫 0、1 分布。例如扔硬币，要么正面，要么反面。
    // 具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要
    // 第一个参数：抽取的数据是否放回，false：不放回
    // 第二个参数：抽取的几率，范围在[0,1]之间,0：全不取；1：全取；
    // 第三个参数：随机数种子
    val dataRDD1 = dataRDD.sample(false, 0.5)
    // 抽取数据放回（泊松算法）
    // 第一个参数：抽取的数据是否放回，true：放回；false：不放回
    // 第二个参数：重复数据的几率，范围大于等于 0.表示每一个元素被期望抽取到的次数
    // 第三个参数：随机数种子
    val dataRDD2 = dataRDD.sample(true, 2)

1.9 distinct

val dataRDD = sparkContext.makeRDD(List(1,2,3,4,1,2),1)
val dataRDD1 = dataRDD.distinct()
val dataRDD2 = dataRDD.distinct(2)

1.10 coalesce

注意: coalesce 默认不会打乱分区中的数据. 缩减分区主要是单纯的进行分区间的合并, 为了避免可能的数据倾斜, 此方法的参数 shuffle = true, 通过shuffle去平衡数据;

val dataRDD = sparkContext.makeRDD(List(1,2,3,4,1,2),6)
val dataRDD1 = dataRDD.coalesce(2)

Q: 如何扩大分区??

• 其实, 直接用下面的这个方法就可以了.

1.11 repartition

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EVOB3uZv-1646825459606)(2022-03-08-16-45-38.png)]

coalesce 和 repartition 区别？

1.12 sortBy

2. 双Value类型

2.13 intersection

2.14 union

2.15 subtract

2.16 zip

 def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)

    // TODO 算子 - 双Value类型

    // 交集，并集和差集要求两个数据源数据类型保持一致
    // 拉链操作两个数据源的类型可以不一致

    val rdd1 = sc.makeRDD(List(1,2,3,4))
    val rdd2 = sc.makeRDD(List(3,4,5,6))
    val rdd7 = sc.makeRDD(List("3","4","5","6"))

    // 交集 : 【3，4】
    val rdd3: RDD[Int] = rdd1.intersection(rdd2)
    //val rdd8 = rdd1.intersection(rdd7)
    println(rdd3.collect().mkString(","))

    // 并集 : 【1，2，3，4，3，4，5，6】
    val rdd4: RDD[Int] = rdd1.union(rdd2)
    println(rdd4.collect().mkString(","))

    // 差集 : 【1，2】
    val rdd5: RDD[Int] = rdd1.subtract(rdd2)
    println(rdd5.collect().mkString(","))

    // 拉链 : 【1-3，2-4，3-5，4-6】
    val rdd6: RDD[(Int, Int)] = rdd1.zip(rdd2)
    val rdd8 = rdd1.zip(rdd7)
    println(rdd6.collect().mkString(","))

    sc.stop()
  }

思考一个问题：如果两个 RDD 数据类型不一致怎么办？
思考一个问题：如果两个 RDD 数据分区不一致怎么办？
思考一个问题：如果两个 RDD 分区数据数量不一致怎么办？

1. 交集，并集和差集要求两个数据源数据类型保持一致
2. 拉链操作两个数据源的类型可以不一致
3. [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eyLK105S-1646825459608)(2022-03-08-17-00-31.png)]

3. Key-Value 类型

3.17 partitionBy

• 思考一个问题：如果重分区的分区器和当前 RDD 的分区器一样怎么办？

  • 产生的结果都是一样的

• 思考一个问题：Spark 还有其他分区器吗？
  

  • RangePartitioner 一般在排序中使用
    思考一个问题：如果想按照自己的方法进行数据分区怎么办？

• 自己写一个分区器(待补充)
  思考一个问题：哪那么多问题？

3.18 reduceByKey

val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
val dataRDD2 = dataRDD1.reduceByKey(_ _)
val dataRDD3 = dataRDD1.reduceByKey(_ _, 2)

Q: redeceByKey 和 groupBykey的区别?

从shuffle 的角度：reduceByKey 和 groupByKey 都存在 shuffle 的操作，但是 reduceByKey可以在 shuffle 前对分区内相同 key 的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而 groupByKey 只是进行分组，不存在数据量减少的问题，reduceByKey 性能比较高。
从功能的角度：reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组，不能聚合，所以在分组聚合的场合下，推荐使用 reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用 groupByKey

3.19 groupByKey

val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
val dataRDD2 = dataRDD1.groupByKey()
val dataRDD3 = dataRDD1.groupByKey(2)
val dataRDD4 = dataRDD1.groupByKey(new HashPartitioner(2))

3.20 aggregateByKey

思考一个问题：分区内计算规则和分区间计算规则相同怎么办？

3.21 foldByKey

3.22 combineByKey

3.23 sortByKey

3.24 join

3.25 leftOuterJoin

3.26 cogroup

5.1.4.4 RDD 行动算子 (Action)

1. reduce

2. collect

3. count

4. first

5. take

6. takeOrdered

7. aggregate

8. fold

9. countByKey

10. save相关的算子

11. foreach

这篇好文章是转载于：学新通技术网