【RDD Transformation】map、filter、flatMap、sample

╰+哭是因爲堅強的太久メ 2022-11-25 13:13 248阅读 0赞

# 一、rdd转化算子 #

1、【map】对Rdd中的数据进行遍历，并根据传入的函数对每个数据进行相关操作；

2、【filter】过滤出rdd中想要的数据；

3、【flatMap】是对rdd中的数据进行一个压平处理；

4、【sample】随机抽样，参数一：true放回，false不放回；参数二：抽样比例，浮点值，是个浮动值，但不是准确值，50%即是0.5；参数三：种子，默认值；

# 二、实例 #

package com.cn.rddOperator
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Transformation01 {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("transformation01").setMaster("local[1]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        //创建一个rdd
        val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6))
        /**
          * map对Rdd中的数据进行遍历，并根据传入的函数对每个数据进行相关操作；
          * collect为action算子，将rdd转化为数组或者集合的形式并展示所有数据；
          * collect: 收集一个弹性分布式数据集的所有元素到一个数组中,这样便于我们观察
          * 会从远程集群拉取数据到driver端。最后，将大量数据
          * 汇集到一个driver节点上，将数据用数组存放，占用了jvm堆内存，非常用意造成内存溢出，只用作小型数据的观察
          **/
        val rdd01: RDD[Int] = rdd.map(x => x * 2)
        println(rdd01.collect().toBuffer)//ArrayBuffer(2, 4, 6, 8, 10, 12)
        /**
          * filter：过滤出rdd中想要的数据
          */
        val rdd02: RDD[Int] = rdd.filter(x => x % 2 == 0)
        println(rdd02.collect().toBuffer)//ArrayBuffer(2, 4, 6)
        /**
          * flatMap是对rdd中的数据进行一个压平处理
          */
        val rdd03: RDD[String] = sc.parallelize(Array("spark java hadoop", "kafka hbase es"))
        val rdd04: RDD[String] = rdd03.flatMap(x => x.split(" "))
        // val rdd05: RDD[Array[String]] = rdd03.map(_.split(" "))
        println(rdd04.collect().toList)//List(spark, java, hadoop, kafka, hbase, es)
    
        /**
          * 随机抽样sample
          * 参数一：true放回，false不放回
          * 参数二：抽样比例，浮点值，是个浮动值，但不是准确值，50%即是0.5;
          * 参数三：种子，默认值
          */
        val rdd06: RDD[Int] = rdd.sample(true,0.5)
        println(rdd06.collect().toBuffer)//ArrayBuffer(1, 1, 2, 3, 5, 6)
      }
    }