当前位置： 首页 >> Spark >> Spark RDD转换算子-单Value类型

---

Spark RDD转换算子-单Value类型

2023-09-10 21:58:29 星期日  发表于北京  阅读：1012

---

 RDD中的所有转换都是延迟加载的，也就是说，它们并不会直接计算结果。相反的，它们只是记住这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给Driver的动作时，这些转换才会真正运行。这种设计让Spark更加有效率地运行 #### Spark的转换与操作（行动） |序号|类型|解释|示例| |-|-|-|-| |1|转换|功能的补充和封装，也就是把一个旧的RDD转换成一个新的RDD|map与flatMap等函数| |2|操作|触发任务调度和作业的执行|collect等函数| 转换和操作都叫做RDD的方法，行业内一般叫做RDD算子。所以RDD的算子分为两种： >转换算子 操作算子 #### 转换算子类型 RDD 根据数据处理方式的不同将算子整体上分为 `单 Value 类型` `双 Value 类型` `Key-Value 类型` 本节介绍单Value类型算子。**所谓单value类型，就是算子只能接收一个数据源（RDD）** ##### map算子 将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。 ``` package cn.libins.learn.spark.core.rdd import org.apache.spark.{SparkConf, SparkContext} object RDDOperatorTransform {   def main(args: Array[String]): Unit = {     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")     val sc = new SparkContext(sparkConf)     val rdd = sc.makeRDD(List(1, 2, 3, 4))     // rdd中的每个元素*2     val transRdd = rdd.map((num: Int) => {       num * 2     })     transRdd.collect().foreach(println)     sc.stop()   } } ``` ##### map算子的并行计算 1、RDD一个分区内的数据是按照读取顺序执行的（串行），如果处理数据的算子之间有依赖，则可以放到一个分区内执行 2、RDD多个分区的数据执行是并行执行的（无序） mapPartitions算子 一次性把RDD一个分区的数据拿过来再计算，提高计算的性能。但是也带来一个问题，因为计算完后内存不会释放（因为存在对象引用），如果内存不够的话，可能会导致OOM ``` val rdd = sc.makeRDD(List(1, 2, 3, 4)) val mapRdd = rdd.mapPartitions(   iter => {     iter.map((num: Int) => {       num * 2     })   } ) mapRdd.collect().foreach(println) ``` ##### 简单总结map 和 mapPartitions 的区别 数据处理角度 Map 算子是分区内一个数据一个数据的执行，类似于串行操作。而 mapPartitions 算子 是以分区为单位进行批处理操作。 功能的角度 Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。 MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变， 所以可以增加或减少数据 性能的角度 Map 算子因为类似于串行操作，所以性能比较低，而是 mapPartitions 算子类似于批处理，所以性能较高。但是 mapPartitions 算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。 ##### mapPartitionsWithIndex算子 将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。（ 应用场景之一：只处理特定分区的数据 ``` package cn.libins.learn.spark.core.rdd import org.apache.spark.{SparkConf, SparkContext} object RDDMapPartitionsWithIndex {   def main(args: Array[String]): Unit = {     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")     val sc = new SparkContext(sparkConf)     val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8))     val mapRdd = rdd.mapPartitionsWithIndex(       (index, data) => {         if (index == 1) { // 只处理第二个分区的数据（索引为1）           data.map(num => {             (index, num) // 返回当前数据在哪个分区           })         } else {           Nil.iterator         }       }     )     mapRdd.collect().foreach(println)     sc.stop()   } } ``` ##### flatMap算子 将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射。 示例：将输入多行转换为一行 ``` val rdd = sc.makeRDD(List("Hello Spark", "Hello Scala", "Hello Hadoop")) val mapRdd = rdd.flatMap(   s => {     s.split(" ")   } ) mapRdd.collect().foreach(println) ``` ##### glom算子 将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变（转换前后分区数量不变，数据转换之后的所在的分区也不会发生变化，因为分区之间的数据是隔离的）。功能有点类似于flatMap的逆操作。 flatMap：把整体拆成个体 glom：把个体封装成整体 示例：计算所有分区最大值求和(分区内取最大值，分区间最大值求和) ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为2） val rdd = sc.makeRDD(List(1, 2, 3, 4), 2) val glomRdd = rdd.glom() // 获取两个分区的最大值 val maxRdd = glomRdd.map(array => {array.max}) // 将两个最大值进行相加并打印 println(maxRdd.collect().sum) ``` ##### groupBy算子 将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为 shuffle。极限情况下，数据可能被分在同一个分区中一个组的数据在一个分区中，但是并不是说一个分区中只有一个组。   示例：将输入数据按照奇偶分组 ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为2） val rdd = sc.makeRDD(List(1, 2, 3, 4), 2) def groupFunction(num: Int): Int = {   num % 2 // 模2有余数，则表示奇数，否则是偶数 } // groupBy调用自定义group by函数 val groupRdd = rdd.groupBy(groupFunction) groupRdd.collect().foreach(println) 输出结果 (0,CompactBuffer(2, 4)) (1,CompactBuffer(1, 3)) ``` 注意：分组和分区没有必然关系，有可能所有的分组都在一个分区里的 ##### filter算子 将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。 当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。 ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为2） val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)   // 只取偶数 val filterRdd = rdd.filter(num => {num % 2 == 0}) filterRdd.collect().foreach(println) ``` ##### 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、可在数据倾斜的时候使用该算子（shuffle打乱后，可能会将大多数数据放到一个分区里） 2、抽奖 ##### distinct算子 将数据集中重复的数据去重 ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为2） val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 4, 4, 2, 3), 2) val distinctRdd = rdd.distinct() distinctRdd.collect().foreach(println) ``` ##### coalesce算子 根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率 当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少分区的个数，减小任务调度成本。 ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为4） val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8), 4) // 第一个参数：合并为2个分区 // 第二个参数：true表示对合并后的分区进行shuffle操作，防止数据再次倾斜 val coaRdd = rdd.coalesce(2, true) coaRdd.saveAsTextFile("data/output/coalesce") ``` 这种情况下的缩减分区，又可能会导致分区数据不均衡，解决方法是：再进行shuffle操作，直接将coalesce算子的第二个参数改为true即可。 生成的文件为 ##### repartition算子 该操作内部其实执行的是 coalesce 操作，参数 shuffle 的默认值为 true。无论是将分区数多的 RDD 转换为分区数少的 RDD，还是将分区数少的 RDD 转换为分区数多的 RDD，repartition 操作都可以完成，因为无论如何都会经 shuffle 过程。 也就是说，repartition算子底层实际上是调用coalesce算子的，你直接调用coalesce算子，并将shuffle参数设置为true也能达到相同的效果。 示例：将输入数据的2个分区，括大为4个分区 ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为2） val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8), 2) val repartRdd = rdd.repartition(4) //  等价于: val repartRdd = rdd.coalesce(4, true) repartRdd.saveAsTextFile("data/output/repartition") ``` 注意：因为经过shuffle，所以每个分区里的数据是随机组装的 ##### sortBy算子 该操作用于排序数据。在排序之前，可以将数据通过 f 函数进行处理，之后按照 f 函数处理 的结果进行排序，默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一 致。中间存在 shuffle 的过程 示例： 将给定的元祖，按照key来降序排序（key是字符串类型，需要转为int类型再排序） ``` // 根据所给元素创建一个RDD（并指定该RDD的分区数为2） val rdd = sc.makeRDD(List(("12", 5), ("3", 6), ("19", 8)), 2) //第一个参数：按照输入元组的第一个元素转为int后的值作为排序条件 //第二个参数：false表示降序排列，默认为升序排列 val sortRdd = rdd.sortBy(_._1.toInt, false) sortRdd.collect().foreach(println) 输出结果 (19,8) (12,5) (3,6) ```