• spark-RDD的两大方法

     RDD方法又称为RDD算子

    RDD转换算子

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

    一、单Value类型（一个数据源）

    1. map函数

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

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

    示例：

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd = sc.makeRDD(List(1, 2, 3, 4))
    13.     // 1,2,3,4 => 2,4,6,8
    14.  
    15.     def mapFunction(num:Int): Int = {
    16.       num * 2
    17.     }
    18.     //转换算子，将一个旧的RDD加上(num * 2)获得一个新的RDD
    19.     val mapRDD: RDD[Int] = rdd.map(mapFunction)
    20.  
    21.     mapRDD.collect().foreach(println)
    22.     sc.stop()
    23.   }
    24.  
    25. }

    1.1 小功能：从服务器日志数据 apache.log 中获取用户请求URL 资源路径

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark_RDD_Operator_Transform_Test {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd = sc.textFile("datas/apache.log")
    13.  
    14.     //长字符串 =>短字符串
    15.     val mapRDD: RDD[String] = rdd.map(
    16.       line => {
    17.         val datas = line.split(" ")
    18.         datas(6)
    19.       }
    20.     )
    21.     mapRDD.collect().foreach(println)
    22.  
    23.     sc.stop()
    24.   }
    25.  
    26. }

    补充：分区内有序执行，分区外无序执行。

    2. mapPartitions函数

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark02_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd = sc.makeRDD(List(1,2,3,4),2)
    13.  
    14.     //mapPartitions可以以分区为单位进行转换操作
    15.     //但是会将整个分区的数据加载到内存进行引用
    16.     //如果处理完的数据是不会被释放掉，存在对象的引用。
    17.     //在内存较小，数据量较大的场合下，容易出现内存溢出。
    18.     val mpRDD: RDD[Int] = rdd.mapPartitions(
    19.       iter => {
    20.         println(">>>>>>>>>>>>>")
    21.         iter.map(_ * 2)
    22.  
    23.       }
    24.     )
    25.  
    26.     mpRDD.collect().foreach(println)
    27.     sc.stop()
    28.   }
    29.  
    30. }

    函数签名：

    def mapPartitions[U: ClassTag]( f: Iterator[T] => Iterator[U],

    preservesPartitioning: Boolean = false): RDD[U]

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

     2.1 获取每个数据分区的最大值。

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark02_RDD_Operator_Transform_test {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd = sc.makeRDD(List(1,2,3,4),2)
    13.     //[1,2],[3,4]两个分区
    14.     //mapPartitions因为是以分区为单位，所以可以获得每个分区的最大值：[2],[4]
    15.     val mpRDD = rdd.mapPartitions(
    16.       iter => {
    17.         List(iter.max).iterator
    18.       }
    19.     )
    20.     mpRDD.collect().foreach(println)
    21.  
    22.     sc.stop()
    23.   }
    24.  
    25. }

    2.2 map和mapPartitions的区别：

    • 数据处理角度：Map 算子是分区内一个数据一个数据的执行，类似于串行操作。而 mapPartitions 算子是以分区为单位进行批处理操作。
    • 功能角度：Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变， 所以可以增加或减少数据。
    • 性能角度：

      Map 算子因为类似于串行操作，所以性能比较低，而是 mapPartitions 算子类似于批处理，所以性能较高。但是mapPartitions 算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用 map 操作。

    3. mapPartitionsWithIndex函数

    函数签名：

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

    preservesPartitioning: Boolean = false): RDD[U]

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

     3.1 小功能：获取第二个数据分区的数据

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark03_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd = sc.makeRDD(List(1,2,3,4),2)
    13.     //[1,2],[3,4],保留第二个数据
    14.     val mpiRDD = rdd.mapPartitionsWithIndex(
    15.       (index,iter) => {
    16.         if(index == 1) {
    17.           iter
    18.         } else {
    19.           Nil.iterator
    20.         }
    21.       }
    22.     )
    23.     mpiRDD.collect().foreach(println)
    24.  
    25.     sc.stop()
    26.   }
    27.  
    28. }

    4. flatMap函数

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark04_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd:RDD[List[Int]] = sc.makeRDD(List(
    13.       List(1,2),List(3,4)
    14.     ))
    15.     //想要的最终结果是把1,2,3,4独立拿出来使用
    16.  
    17.     val flatRDD: RDD[Int] = rdd.flatMap(
    18.       list => {
    19.         list
    20.       }
    21.     )
    22.     flatRDD.collect().foreach(println)
    23.     sc.stop()
    24.   }
    25.  
    26. }

    4.1 小功能：将 List(List(1,2),3,List(4,5))进行扁平化操作

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark04_RDD_Operator_Transform1 {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = map
    12.     val rdd = sc.makeRDD(List(List(1,2),3,List(4,5)))
    13.     //小功能：将 List(List(1,2),3,List(4,5))进行扁平化操作
    14.     //因为数据有不同的类型，进行模式匹配：case
    15.  
    16.     val flatRDD = rdd.flatMap(
    17.       data => {
    18.         data match {
    19.           case list: List[_] => list
    20.           case dat => List(dat)
    21.         }
    22.       }
    23.     )
    24.     flatRDD.collect().foreach(println)
    25.     sc.stop()
    26.   }
    27.  
    28. }

    5. glom方法

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark05_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = glom
    12.     val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
    13.     //list => int
    14.     //int => array
    15.     val glomRDD: RDD[Array[Int]] = rdd.glom()
    16.  
    17.     glomRDD.collect().foreach(data => println(data.mkString(",")))
    18.     //将一个分区的数据形成一个数组
    19.     sc.stop()
    20.   }
    21.  
    22. }

    5.1 小功能：计算所有分区最大值求和（分区内取最大值，分区间最大值求和）

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark05_RDD_Operator_Transform_test {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = glom
    12.     val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
    13.     //[1,2],[3,4]
    14.     //计算所有分区最大值求和（分区内取最大值，分区间最大值求和）
    15.     //最大值：[2],[4],
    16.     // 分区间求和：[6]
    17.     val glomRDD: RDD[Array[Int]] = rdd.glom()
    18.  
    19.     val maxRDD: RDD[Int] = glomRDD.map(
    20.       Array => {
    21.         Array.max
    22.       }
    23.     )
    24.  
    25.     println(maxRDD.collect().sum)
    26.     //将一个分区的数据形成一个数组
    27.     sc.stop()
    28.   }
    29.  
    30. }

    6. groupBy方法

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark06_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = groupBy
    12.     //实现分组
    13.     val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
    14.     /*
    15.     groupBy会将数据源中的每一个数据进行分组判断，根据返回的分组key进行分组
    16.     相同的key值的数据会放置在同一个组中
    17.      */
    18.     def groupFunction(num:Int) = {
    19.       num % 2
    20.     }
    21.  
    22.     val groupRDD: RDD[(Int,Iterable[Int])] = rdd.groupBy(groupFunction)
    23.  
    24.     groupRDD.collect().foreach(println)
    25.     sc.stop()
    26.   }
    27.  
    28. }
    29. /*
    30. 结果：
    31. (0,CompactBuffer(2, 4))
    32. (1,CompactBuffer(1, 3))
    33. 余数是0的放在一起，余数是1的放在一起。
    34.  */

    6.1 小功能：将 List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark06_RDD_Operator_Transform_1 {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = groupBy
    12.     //实现分组
    13.     val rdd = sc.makeRDD(List("Hello", "hive", "hbase", "Hadoop"),2)
    14.     /*
    15.     groupBy会将数据源中的每一个数据进行分组判断，根据返回的分组key进行分组
    16.     相同的key值的数据会放置在同一个组中
    17.      */
    18.  
    19.     val groupRDD = rdd.groupBy(_.charAt(0))
    20.  
    21.     groupRDD.collect().foreach(println)
    22.     sc.stop()
    23.   }
    24.  
    25. }
    26. /*
    27. 结果：
    28. (h,CompactBuffer(hive, hbase))
    29. (H,CompactBuffer(Hello, Hadoop))
    30.  */

    7. filter函数

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark07_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = filter
    12.     //实现分组
    13.     val rdd = sc.makeRDD(List(1,2,3,4))
    14.     //实现：保留奇数，舍去偶数
    15.     val filterRDD = rdd.filter(num => num % 2 != 0)
    16.     
    17.     filterRDD.collect().foreach(println)
    18.     sc.stop()
    19.   }
    20.  
    21. }
    22. /*
    23. 结果：
    24. 1
    25. 3
    26.  */

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark07_RDD_Operator_Transform_test {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = filter
    11.     //实现分组
    12.     val rdd = sc.textFile("datas/apache.log")
    13.     //从服务器日志数据 apache.log 中获取 2015 年 5 月 17 日的请求路径
    14.     rdd.filter(
    15.       line => {
    16.         val datas = line.split(" ")
    17.         datas
    18.         val time = datas(3)
    19.         time.startsWith("17/05/2015")
    20.       }
    21.     ).collect().foreach(println)
    22.  
    23.     sc.stop()
    24.   }
    25.  
    26. }

    8. sample函数

    函数签名：

    def sample( withReplacement: Boolean, fraction: Double,

    seed: Long = Utils.random.nextLong): RDD[T]

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark08_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = sample
    11.     //实现分组
    12.     val rdd = sc.makeRDD(List(1,2,3,4,5,6,7,8,9,10))
    13.     //实现：根据指定的规则从数据集中抽取数据
    14.     /*
    15.     sample有三个参数：
    16.     1. 表示抽取数据是否放回
    17.     2.表示每条数据可能被抽取的概率
    18.       如果抽取不放回，该参数代表数据源中每条数据被抽取的概率
    19.       如果抽取放回，该参数代表数据源中的每条数据被抽取的可能次数
    20.       2.1 基准值的概念：
    21.     3.表示抽取的数据的随机算法的种子
    22.       3.1 如果不传递第三个参数，那么使用的是当前系统时间
    23.      */
    24.     println(rdd.sample(
    25.       true,
    26.       //0.4,
    27.       1
    28.  
    29.     ).collect().mkString(","))
    30.  
    31.     sc.stop()
    32.   }
    33. }
    34. /*
    35. 结果：
    36. 4,8,8,9
    37. 多次运行会发现，该随机数是伪随机，就是多次的结果一样
    38. 也就是第二个参数确定，则随机种子就确定，值确定
    39.  */

    用途：来检验数据倾斜是由于哪些数据造成的。

    9. distinct函数

    函数签名：

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

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark09_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = sample
    12.     //实现分组
    13.     val rdd = sc.makeRDD(List(1,2,3,4,1,2,3,4))
    14.     //实现：将数据集中重复的数据去重
    15.     val rdd1: RDD[Int] = rdd.distinct()
    16.     rdd1.collect().foreach(println)
    17.  
    18.     sc.stop()
    19.   }
    20. }
    21.  
    22. /*
    23. 结果：
    24. 1,2,3,4
    25.  */

    10. coalesce函数

    函数签名：

    def coalesce(numPartitions: Int, shuffle: Boolean = false,

    partitionCoalescer: Option[PartitionCoalescer] = Option.empty) (implicit ord: Ordering[T] = null)

    : RDD[T]

    函数说明：根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率，当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少分区的个数，减小任务调度成本。

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark10_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = coalesce
    12.     //根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率
    13.     val rdd = sc.makeRDD(List(1,2,3,4,5,6),3)
    14.     //实现：缩减分区
    15.     /*
    16.     coalesce方法默认情况下不会将分区的数据打乱重新组合。
    17.     也就是相同分区的数据不会拆开，依然放到同一个分区里
    18.     这种情况下缩减分区可能会导致数据不均衡，出现数据倾斜
    19.     为了防止这种问题，我们可以进行shuffle处理,也就是将第二个参数改成true
    20.      */
    21.     val newRDD: RDD[Int] = rdd.coalesce(2,true)
    22.     newRDD.saveAsTextFile("output")
    23.  
    24.     sc.stop()
    25.   }
    26. }
    27.  
    28. /*
    29. 结果：
    30.  */

    11. reparation函数

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark11_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = repartition
    12.     //根据数据量改变分区
    13.     val rdd = sc.makeRDD(List(1,2,3,4,5,6),2)
    14.     //实现：扩大分区
    15.     /*
    16.     底层代码就是用的coalesce
    17.      */
    18.     val newRDD: RDD[Int] = rdd.repartition(3)
    19.     newRDD.saveAsTextFile("output")
    20.  
    21.     sc.stop()
    22.   }
    23. }
    24.  
    25. /*
    26. 结果：
    27.  */

    12. sortBy

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark12_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = sortBy
    12.     //数据排序,全局有序，所有数据先都拿出来排序，排完在分区
    13.     val rdd = sc.makeRDD(List(6,5,4,3,2,1),2)
    14.     val newRDD: RDD[Int] = rdd.sortBy(num => num)
    15.     newRDD.saveAsTextFile("output")
    16.     sc.stop()
    17.   }
    18. }
    19.  
    20. /*
    21. 结果：
    22.  */

    二、双Value类型（两个数据源）

    13. intersection函数（交集）

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

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

    14. union函数（并集）

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

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

    15. subtract函数（差集）

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

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

    16. zip函数（拉链）

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. object Spark13_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = 双value类型
    12.     val rdd1 = sc.makeRDD(List(1,2,3,4))
    13.     val rdd2 = sc.makeRDD(List(3,4,5,6))
    14.  
    15.     //交集 [3,4]
    16.     val rdd3: RDD[Int] = rdd1.intersection(rdd2)
    17.     println(rdd3.collect().mkString(","))
    18.     //并集 [1,2,3,4,3,4,5,6]
    19.     val rdd4: RDD[Int] = rdd1.union(rdd2)
    20.     println(rdd4.collect().mkString(","))
    21.     //差集 [1,2]
    22.     val rdd5: RDD[Int] = rdd1.subtract(rdd2)
    23.     println(rdd5.collect().mkString(","))
    24.     //拉链 [1-3,2-4,3-5,4-6]
    25.     val rdd6: RDD[(Int,Int)] = rdd1.zip(rdd2)
    26.     println(rdd6.collect().mkString(","))
    27.  
    28.     sc.stop()
    29.   }
    30. }
    31.  
    32. /*
    33. 结果：
    34.  */
    • 交集并集差集要求两个数据源的数据类型保存一致。
    • 拉链操作两个数据源类型可以不一致，但要求两个数据源的分区数量保持一致，并且每一个分区数据数量要相同。

    三、Key-Value类型

    17. partitionBy函数

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
    5.  
    6. object Spark17_RDD_Operator_Transform {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 算子 = Key-Value类型
    12.     val rdd = sc.makeRDD(List(1,2,3,4),2)
    13.     val mapRDD = rdd.map((_,1))
    14.     //partitionBy根据指定的分区规则对数据进行重分区
    15.     mapRDD.partitionBy(new HashPartitioner(2)).saveAsTextFile("output")
    16.     //该方法是在PairRDDFunctions里面的
    17.     //隐式转换（二次编译），将一个类型变成另外一个类型
    18.     sc.stop()
    19.   }
    20. }
    21.  
    22. /*
    23. 结果：
    24. (2,1)
    25. (4,1)
    27. (1,1)
    28. (3,1)
    29.  */

    18. reduceByKey

    函数签名：def reduceByKey(func: (V, V) => V): RDD[(K, V)]

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
    4.  
    5. object Spark18_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型
    11.     val rdd = sc.makeRDD(List(
    12.       ("a",1),("a",2),("a",3),("b",4)
    13.     ))
    14.  
    15.     //reduceByKey:相同的Key进行value操作
    16.     //scala中的菊科一般都是两两聚合，spark基于scala开发的，所以它也是两两聚合
    17.     val reduceRDD = rdd.reduceByKey((x: Int, y: Int) => {
    18.       x + y
    19.     })
    20.     //reduceByKey中如果key的数据只有一个，则不会参与运算。
    21.     reduceRDD.collect().foreach(println)
    22.  
    23.     sc.stop()
    24.   }
    25. }
    26.  
    27. /*
    28. 结果：
    29. (a,6)
    30. (b,4)
    31.  */

    19. groupByKey函数

    函数说明：将数据源的数据根据 key 对 value 进行分组。

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark19_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型
    11.     val rdd = sc.makeRDD(List(
    12.       ("a",1),("a",2),("a",3),("b",4)
    13.     ))
    14.  
    15.     //groupByKey:将相同数据源中的数据，相同key的数据放在一个组中，形成一个对偶元组
    16.     //          元组中的第一个元素就是key，第二个元素就是相同key的value的集合
    17.     val groupRDD = rdd.groupByKey()
    18.     groupRDD.collect().foreach(println)
    19.  
    20.     sc.stop()
    21.   }
    22. }
    23.  
    24. /*
    25. 结果：
    26. (a,CompactBuffer(1, 2, 3))
    27. (b,CompactBuffer(4))
    28.  */

    19.1 reduceByKey 和 groupByKey 的区别？

    • 从 shuffle 的角度：reduceByKey 和 groupByKey 都存在 shuffle 的操作，但是reduceByKey 可以在 shuffle 前对分区内相同 key 的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而groupByKey 只是进行分组，不存在数据量减少的问题，reduceByKey 性能比较高。

    • spark中，shuffle操作必须落盘处理（就是存到文件当中进行等待），不能在内存中进行数据等待，会导致内存溢出。

    • 从功能的角度：reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组，不能聚合，所以在分组聚合的场合下，推荐使用 reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey。

    20. aggregateByKey函数

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark20_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-aggregateByKey
    11.     val rdd = sc.makeRDD(List(
    12.       ("a",1),("a",2),("a",3),("a",4)
    13.     ),2)
    14.     //分区内相同的key求最大值，分区间相同的key求和
    15.     //aggregateByKey存在参数的柯里化
    16.     //第一个参数列表：需要传递一个参数，表示为初始值
    17.     //    主要用于当碰见第一个key的时候，和value进行分区内的计算
    18.     //第二个参数列表：需要传递两个参数
    19.     //    第一个参数表示分区内计算规则
    20.     //    第二个参数表示分区间计算规则
    21.     rdd.aggregateByKey(0)(
    22.       (x,y) => math.max(x,y),
    23.       (x,y) => x+y
    24.     ).collect().foreach(println)
    25.  
    26.     sc.stop()
    27.   }
    28. }
    29.  
    30. /*
    31. 结果：
    32. (a,6)
    33.  */

    20.1 求数据平均值

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark20_RDD_Operator_Transform_test {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-aggregateByKey
    11.     val rdd = sc.makeRDD(List(
    12.       ("a",1),("a",2),("b",3),
    13.       ("b",4),("b",5),("a",6)
    14.     ),2)
    15.     //获取相同key的数据的平均值
    16.     val newRDD = rdd.aggregateByKey((0, 0))(
    17.       (t, v) => {
    18.         (t._1 + v, t._2 + 1)
    19.       },
    20.       (t1, t2) => {
    21.         (t1._1 + t2._1, t1._2 + t2._2)
    22.       }
    23.     )
    24.  
    25.     val resRDD = newRDD.mapValues {
    26.       case (num, cnt) => {
    27.         num / cnt
    28.       }
    29.     }
    30.     resRDD.collect().foreach(println)
    31.     sc.stop()
    32.   }
    33. }
    34.  
    35. /*
    36. 结果：
    37. (b,4)
    38. (a,3)
    39.  */

    21. foldByKey函数

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark21_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-foldByKey
    11.     val rdd = sc.makeRDD(List(
    12.       ("a",1),("a",2),("b",3),
    13.       ("b",4),("b",5),("a",6)
    14.     ),2)
    15.     //如果聚合计算时，分区内和分区间计算规则相同，spark提供了简化的方法foldByKey
    16.     rdd.foldByKey(0)(_+_).collect().foreach(println)
    17.     sc.stop()
    18.   }
    19. }
    20.  
    21. /*
    22. 结果：
    23. (b,12)
    24. (a,9)
    25.  */

    22. combineByKey函数

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark22_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-combineByKey
    11.     val rdd = sc.makeRDD(List(
    12.       ("a",1),("a",2),("b",3),
    13.       ("b",4),("b",5),("a",6)
    14.     ),2)
    15.     //获取相同key的数据的平均值
    16.     /*
    17.     combineByKey方法需要三个参数:
    18.     第一个参数：将相同key的第一个数据进行结构的转换，实现操作
    19.     第二个参数：分区内的计算规则
    20.     第三个参数：分区间的计算规则
    21.      */
    22.     val newRDD = rdd.combineByKey(
    23.       v =>(v,1),
    24.       (t:(Int,Int), v) => {
    25.         (t._1 + v, t._2 + 1)
    26.       },
    27.       (t1:(Int,Int), t2:(Int,Int)) => {
    28.         (t1._1 + t2._1, t1._2 + t2._2)
    29.       }
    30.     )
    31.  
    32.     val resRDD = newRDD.mapValues {
    33.       case (num, cnt) => {
    34.         num / cnt
    35.       }
    36.     }
    37.     resRDD.collect().foreach(println)
    38.     sc.stop()
    39.   }
    40. }
    41.  
    42. /*
    43. 结果：
    44. (b,4)
    45. (a,3)
    46.  */

    思考：reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别

    • reduceByKey: 相同 key 的第一个数据不进行任何计算，分区内和分区间计算规则相同。

    • FoldByKey: 相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同。

    • AggregateByKey：相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同。

    • CombineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区内和分区间计算规则不相同。

    24. join函数

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

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

    将两个数据源中的数据相同的key连接在一起。

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark24_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-combineByKey
    11.     val rdd1 = sc.makeRDD(List(
    12.       ("a",1),("b",2),("c",3)
    13.     ))
    14.     val rdd2 = sc.makeRDD(List(
    15.       ("a",4),("b",5),("c",6)
    16.     ))
    17.     /*
    18.     join:
    19.       两个不同数据源的数据，相同的key的value会连接在一起，形成元组
    20.       如果两个数据源中key没有匹配上，那么数据不会出现在结果中
    21.       如果两个数据源中key有多个相同的，会依次匹配，可能会出现笛卡尔乘积，数据量会几何增长，会导致性能降低
    22.       谨慎使用
    23.      */
    24.     val resRDD = rdd1.join(rdd2)
    25.     resRDD.collect().foreach(println)
    26.     sc.stop()
    27.   }
    28. }
    29.  
    30. /*
    31. 结果：
    32. (a,(1,4))
    33. (b,(2,5))
    34. (c,(3,6))
    35.  */

    25. leftOuterJoin函数

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

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

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark25_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-combineByKey
    11.     val rdd1 = sc.makeRDD(List(
    12.       ("a",1),("b",2),("c",3)
    13.     ))
    14.     val rdd2 = sc.makeRDD(List(
    15.       ("a",4),("b",5)//,("c",6)
    16.     ))
    17.  
    18.     val resRDD = rdd1.leftOuterJoin(rdd2)
    19.     //val rightRDD = rdd1.rightOuterJoin(rdd2)
    20.     resRDD.collect().foreach(println)
    21.     sc.stop()
    22.   }
    23. }
    24.  
    25. /*
    26. 结果：
    27. (a,(1,Some(4)))
    28. (b,(2,Some(5)))
    29. (c,(3,None))
    30.  */

    26. rightOuterJoin函数

    函数说明：类似于 SQL 语句的右外连接。

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark25_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-combineByKey
    11.     val rdd1 = sc.makeRDD(List(
    12.       ("a",1),("b",2)//,("c",3)
    13.     ))
    14.     val rdd2 = sc.makeRDD(List(
    15.       ("a",4),("b",5),("c",6)
    16.     ))
    17.  
    18.     //val resRDD = rdd1.leftOuterJoin(rdd2)
    19.     val rightRDD = rdd1.rightOuterJoin(rdd2)
    20.     rightRDD.collect().foreach(println)
    21.     sc.stop()
    22.   }
    23. }
    24.  
    25. /*
    26. 结果：
    27. (a,(Some(1),4))
    28. (b,(Some(2),5))
    29. (c,(None,6))
    30.  */

    27. cogroup函数

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

     可以实现分组+连接：

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark27_RDD_Operator_Transform {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     // TODO 算子 = Key-Value类型-combineByKey
    11.     val rdd1 = sc.makeRDD(List(
    12.       ("a",1),("b",2)//,("c",3)
    13.     ))
    14.     val rdd2 = sc.makeRDD(List(
    15.       ("a",4),("b",5),("c",6),("c",6)
    16.     ))
    17.     /*
    18.     cogroup:connect+group,连接+分组
    19.      */
    20.     val coRDD = rdd1.cogroup(rdd2)
    21.     coRDD.collect().foreach(println)
    22.     sc.stop()
    23.   }
    24. }
    25.  
    26. /*
    27. 结果：
    28. (a,(CompactBuffer(1),CompactBuffer(4)))
    29. (b,(CompactBuffer(2),CompactBuffer(5)))
    30. (c,(CompactBuffer(),CompactBuffer(6, 6)))
    31.  */

    算子案例实操

    1. package cn.gfs.sparkCore.rdd.operator.transform
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
    5.  
    6. object Spark_final_RDD_Req {
    7.   def main(args: Array[String]): Unit = {
    8.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    9.     val sc = new SparkContext(sparkConf)
    10.  
    11.     // TODO 案例实操
    12.     /*
    13.     数据准备：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
    14.     需求描述：统计出每一个省份每个广告被点击数量排行的 Top3
    15.      */
    16.     //思路设计：
    17.     //1. 获取原始数据
    18.     val dataRDD = sc.textFile("datas/agent.log")
    19.     //2. 将原始数据进行结构的转换，方便统计：((省份，广告)，1)
    20.     val mapRDD = dataRDD.map(
    21.       line => {
    22.         val datas = line.split(" ")
    23.         ((datas(1), datas(4)), 1)
    24.       }
    25.     )
    26.     //3. 将转换后的数据进行分组聚合：((省份，广告)，sun)
    27.     val reduceRDD: RDD[((String,String),Int)] = mapRDD.reduceByKey(_+_)
    28.     //4. 将聚合的结果进行结构的转换：(省份，(广告，sum))
    29.     val newmapRDD = reduceRDD.map{
    30.       case ((prv,ad),sum) =>{
    31.         (prv,(ad,sum))
    32.       }
    33.     }
    34.     //5. 将转换后的数据根据省份进行分组：(省份，((广告A，sum)，(广告B，sum)))
    35.     val groupRDD = newmapRDD.groupByKey()
    36.     //6. 将分组后的数据进行组内排序，降序取前三
    37.     val resultRDD = groupRDD.mapValues(
    38.       iter => {
    39.         iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
    40.       }
    41.     )
    42.     //7. 采集数据，打印控制台
    43.     resultRDD.collect().foreach(println)
    44.  
    45.     sc.stop()
    46.   }
    47. }
    48.  
    49. /*
    50. 结果：
    51. (4,List((12,25), (2,22), (16,22)))
    52. (8,List((2,27), (20,23), (11,22)))
    53. (6,List((16,23), (24,21), (22,20)))
    54. (0,List((2,29), (24,25), (26,24)))
    55. (2,List((6,24), (21,23), (29,20)))
    56. (7,List((16,26), (26,25), (1,23)))
    57. (5,List((14,26), (21,21), (12,21)))
    58. (9,List((1,31), (28,21), (0,20)))
    59. (3,List((14,28), (28,27), (22,25)))
    60. (1,List((3,25), (6,23), (5,22)))
    61.  */

    RDD行动算子

    行动算子就是触发整个计算的执行。

    底层代码调用的是环境对象的runjob方法，底层代码中会创建ActionJob，并提交执行。

    1. reduce方法

    函数签名：def reduce(f: (T, T) => T): T

    函数说明：聚集RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据。（求和）

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark01_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1, 2, 3, 4))
    11.     //TODO - 行动算子： reduce
    12.     val reRDD = rdd.reduce(_ + _)
    13.     println(reRDD)
    14.     sc.stop()
    15.   }
    16.  
    17. }
    18. /*
    19. 结果：
    20. 10
    21.  */

    2. collect方法

    函数签名：def collect(): Array[T]

    函数说明：在驱动程序中，以数组Array 的形式返回数据集的所有元素。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark02_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1, 2, 3, 4))
    11.     //TODO - 行动算子：
    12.  
    13.     //collect:会将不同分区的数据按照分区顺序采集回来，形成数组
    14.     val reRDD = rdd.collect()
    15.     println(reRDD.mkString(","))
    16.     sc.stop()
    17.   }
    18.  
    19. }
    20.  
    21. /*
    22. 结果：
    23. 1,2,3,4
    24.  */

    3. count方法

    函数说明：返回RDD 中元素的个数。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark03_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1, 2, 3, 4))
    11.     //TODO - 行动算子：
    12.  
    13.     val reRDD = rdd.count()
    14.     println(reRDD)
    15.     sc.stop()
    16.   }
    17.  
    18. }
    19.  
    20. /*
    21. 结果：
    22. 4
    23.  */

    4. first方法

    函数说明：返回RDD 中的第一个元素。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark04_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1, 2, 3, 4))
    11.     //TODO - 行动算子：
    12.     //first:获取数据源中的第一个
    13.     val reRDD = rdd.first()
    14.     println(reRDD)
    15.     sc.stop()
    16.   }
    17.  
    18. }
    19.  
    20. /*
    21. 结果：
    22. 1
    23.  */

    5. take方法

    函数说明：返回一个由RDD 的前 n 个元素组成的数组。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark05_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1, 2, 3, 4))
    11.     //TODO - 行动算子：
    12.     //take:取多少个
    13.     val reRDD = rdd.take(3)
    14.     println(reRDD.mkString(","))
    15.     sc.stop()
    16.   }
    17.  
    18. }
    19.  
    20. /*
    21. 结果：
    22. 1,2,3
    23.  */

    6. takeOrdered方法

    函数说明：返回该 RDD 排序后的前 n 个元素组成的数组。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark06_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(4,2,3,1))
    11.     //TODO - 行动算子：
    12.     //takeOrderd:先进行排序，再取多少个
    13.     val reRDD = rdd.takeOrdered(3)
    14.     println(reRDD.mkString(","))
    15.     sc.stop()
    16.   }
    17.  
    18. }
    19.  
    20. /*
    21. 结果：
    22. 1,2,3
    23.  */

    7. aggregate方法

    函数说明：分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark07_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1,2,3,4))
    11.     //TODO - 行动算子：
    12.  
    13.     val reRDD = rdd.aggregate(0)(_+_,_+_)
    14.  
    15.     println(reRDD)
    16.     sc.stop()
    17.   }
    18.  
    19. }
    20.  
    21. /*
    22. 结果：
    23. 10
    24.  */

    8. fold方法

    函数说明：折叠操作，aggregate 的简化版操作。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark08_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     val rdd = sc.makeRDD(List(1,2,3,4),2)
    11.     //TODO - 行动算子：
    12.     //fold:分区内和分区间计算规则相同
    13.     val reRDD = rdd.fold(10)(_+_)
    14.  
    15.     println(reRDD)
    16.     sc.stop()
    17.   }
    18.  
    19. }
    20.  
    21. /*
    22. 结果：
    23. 40
    24.  */

    9. countByKey方法

    函数说明：统计每种 key 的个数。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark09_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     //val rdd = sc.makeRDD(List(1,1,1,4),2)
    11.     //TODO - 行动算子：
    12.     //countByValue:每个值出现几次
    13.     //val reRDD = rdd.countByValue()
    14.  
    15.     //countByKey:
    16.     val rdd = sc.makeRDD(List(("a",1),("a",1),("a",1),("b",1)),2)
    17.     val countRDD = rdd.countByKey()
    18.     println(countRDD)
    19.     sc.stop()
    20.   }
    21.  
    22. }
    23.  
    24. /*
    25. 结果：
    26. countByValue:List(1,2,3,4)
    27. Map(4 -> 1, 2 -> 1, 1 -> 1, 3 -> 1)
    28. countByKey
    29. Map(b -> 1, a -> 3)
    30.  */

    10. save相关算子

    函数说明：将数据保存到不同格式的文件中。

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. object Spark10_RDD_Operator_Action {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    8.     val sc = new SparkContext(sparkConf)
    9.  
    10.     //countByKey:
    11.     val rdd = sc.makeRDD(List(("a",1),("a",1),("a",1),("b",1)),2)
    12.     rdd.saveAsTextFile("output")
    13.     rdd.saveAsObjectFile("output1")
    14.     //saveAsSequenceFile这个方法要求数据的格式必须为k-v类型
    15.     rdd.saveAsSequenceFile("output2")
    16.     //println(countRDD)
    17.     sc.stop()
    18.   }
    19.  
    20. }
    21.  
    22. /*
    23. 结果：
    25.  */

    11. foreach方法

    函数说明： 分布式遍历RDD 中的每一个元素，调用指定函数。

    12. 总结：实现wordcount的几种方法

    1. package cn.gfs.sparkCore.rdd.operator.action
    2.  
    3. import org.apache.spark.{SparkConf, SparkContext}
    4.  
    5. import scala.collection.mutable
    6.  
    7. object Spark_RDD_Operator_WordCount {
    8.   def main(args: Array[String]): Unit = {
    9.     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    10.     val sc = new SparkContext(sparkConf)
    11.  
    12.     wordcount1(sc)
    13.     wordcount2(sc)
    14.     wordcount3(sc)
    15.     wordcount4(sc)
    16.     wordcount5(sc)
    17.     wordcount6(sc)
    18.     wordcount7(sc)
    19.     wordcount8(sc)
    20.     sc.stop()
    21.   }
    22.  
    23.   //groupBy
    24.   def wordcount1(sc : SparkContext):Unit = {
    25.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    26.     val words = rdd.flatMap(_.split(" "))
    27.     val group = words.groupBy(word => word)
    28.     val wordCount = group.mapValues(iter => iter.size)
    29.   }
    30.   //groupByKey
    31.   def wordcount2(sc : SparkContext):Unit = {
    32.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    33.     val words = rdd.flatMap(_.split(" "))
    34.     val wordOne = words.map((_,1))
    35.     val group = wordOne.groupByKey()
    36.     val wordCount = group.mapValues(iter => iter.size)
    37.   }
    38.   //reduceByKey
    39.   def wordcount3(sc : SparkContext):Unit = {
    40.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    41.     val words = rdd.flatMap(_.split(" "))
    42.     val wordOne = words.map((_,1))
    43.     val wordCount = wordOne.reduceByKey(_ + _)
    44.   }
    45.   //aggregateByKey
    46.   def wordcount4(sc : SparkContext):Unit = {
    47.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    48.     val words = rdd.flatMap(_.split(" "))
    49.     val wordOne = words.map((_,1))
    50.     val wordCount = wordOne.aggregateByKey(0)(_+_,_+_)
    51.   }
    52.   //foldByKey
    53.   def wordcount5(sc : SparkContext):Unit = {
    54.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    55.     val words = rdd.flatMap(_.split(" "))
    56.     val wordOne = words.map((_,1))
    57.     val wordCount = wordOne.foldByKey(0)(_ + _)
    58.   }
    59.   //combineByKey
    60.   def wordcount6(sc : SparkContext):Unit = {
    61.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    62.     val words = rdd.flatMap(_.split(" "))
    63.     val wordOne = words.map((_,1))
    64.     val wordCount = wordOne.combineByKey(
    65.       v=>v,
    66.       (x:Int,y)=>x+y,
    67.       (x:Int,y:Int)=>x+y
    68.     )
    69.   }
    70.   //countByKey
    71.   def wordcount7(sc : SparkContext):Unit = {
    72.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    73.     val words = rdd.flatMap(_.split(" "))
    74.     val wordOne = words.map((_,1))
    75.     val wordCount = wordOne.countByKey()
    76.   }
    77.   //countByKey
    78.   def wordcount8(sc : SparkContext):Unit = {
    79.     val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    80.     val words = rdd.flatMap(_.split(" "))
    81.     //val wordOne = words.map((_,1))
    82.     val wordCount = words.countByValue()
    83.   }
    84.   //reduce
    85. //  def wordcount9(sc : SparkContext):Unit = {
    86. //    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    87. //    val words = rdd.flatMap(_.split(" "))
    88. //    val mapWord = words.map(
    89. //      word => {
    90. //        mutable.Map[String, Long]((word, 1))
    91. //      }
    92. //    )
    93. //    //val wordOne = words.map((_,1))
    94. //    val wordCount = mapWord.reduce(
    95. //      (map1,map2) => {
    96. //        map2.foreach{
    97. //          case (word,count)=>{
    98. //            val newCount = map1.getOrElse(word,OL)+count
    99. //            map1.update(word,newCount)
    100. //          }
    101. //        }
    102. //        map1
    103. //      }
    104. //    )
    105. //    println(wordCount)
    106. //  }
    107. }
    108.  
    109. /*
    110. 结果：
    111. )
    112.  */

    算子概念回顾

    算子：Operator（操作）

            RDD的方法和scala集合对象的方法不一样，

            集合对象的方法都是在同一个节点的内存中完成的。

            RDD的方法可以将计算逻辑分发到Executor端（分布式节点）上执行。

            为了区分不同的处理效果，所以将RDD的方法称之为算子。

            RDD的方法外部的操作都是在Driver端执行的，而方法内部的逻辑代码是在Executor端执行。

  • 相关阅读:
    如何从清空的回收站中恢复已删除的Word文档？
    CodeTON Round 6 (Div. 1 + Div. 2) 题解 | JorbanS
    一图读懂「五度情报站」全盘视野，情报智取，先知先行，决策有道！
    jQuery基础----开始简单的了解
    java线程生命周期
    一遍掌握，快速排序的入门理解
    clip:learning transferable visual models from natural language supervision
    Linux 常用基本命令
    Vue 项目前端响应式布局及框架搭建
    el-select
  • 原文地址：https://blog.csdn.net/qq_53914420/article/details/127754640