RDD的转换

柔情只为你懂 2023-07-18 09:58 4阅读 0赞

RDD整体上分为Value类型和Key-Value类型  
**1 Value类型**

//Rdd的转化
      def Test:Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val source=sc.parallelize(1 to 10)//创建一个rdd
        val mapadd = source.map(_*2)
        print(mapadd.collect().toBuffer)
        sc.stop()
      }
      // mapPartitions(func) 案例
      def  Test2:Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd1=sc.parallelize(Array(1,2,3,4))
        val rdd2=rdd1.mapPartitions(x=>x.map(_*2))
        print(rdd2.collect().toBuffer)
        sc.stop()
      }
      //mapPartitionsWithIndex(func) 案例
      def Test3:Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd1=sc.parallelize(Array(1,2,3,4),2)
        val rdd2 = rdd1.mapPartitionsWithIndex((index,items)=>(items.map((index,_))))
        print(rdd2.collect().toBuffer)
        sc.stop()
      }
      //flatMap()1. 作用：类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）
      def Test4():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val sourceFlat = sc.parallelize(1 to 5)
        print(sourceFlat.collect())
        val flatMap = sourceFlat.flatMap(1 to _)
        print(flatMap.collect().toBuffer)
        sc.stop()
    
      }
      /* glom案例
      1. 作用：将每一个分区形成一个数组，形成新的RDD类型时RDD[Array[T]]
      2. 需求：创建一个4个分区的RDD，并将每个分区的数据放到一个数组*/
    
      def Test5():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(1 to 16,4)
        val rdd1=rdd.glom()
        print(rdd1.collect().toBuffer)
        sc.stop()
      }
      /*groupBy(func)案例
        1. 作用：分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。
      2. 需求：创建一个RDD，按照元素模以2的值进行分组。*/
      def Test6():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(1 to 10)
        val group = rdd.groupBy(_%2)
        print(group.collect().toBuffer)
        sc.stop()
      }
     /* filter(func) 案例
        1. 作用：过滤。返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成。
      2. 需求：创建一个RDD（由字符串组成），过滤出一个新RDD（包含”xiao”子串）*/
      def Test7():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        var sourceFilter = sc.parallelize(Array("xiaoming","xiaojiang","xiaohe","dazhi"))
        //过滤出有xiao字的数据
        val filter = sourceFilter.filter(_.contains("xiao"))
        print(filter.collect().toBuffer)
        sc.stop()
      }
      /*sample(withReplacement, fraction, seed) 案例
        1. 作用：以指定的随机种子随机抽样出数量为fraction的数据，withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样，seed用于指定随机数生成器种子。
      2. 需求：创建一个RDD（1-10），从中选择放回和不放回抽样*/
      def Test8():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(1 to 10)
        //放回抽样
        var sample1 = rdd.sample(true,0.4,2)
        print("放回抽样"+sample1.collect().toBuffer)
        //不放回抽样
        var sample2 = rdd.sample(false,0.2,3)
        print("不放回抽样"+sample2.collect().toBuffer)
        sc.stop()
      }
     /* distinct([numTasks])) 案例 说简单点，可以理解为去重
      1. 作用：对源RDD进行去重后返回一个新的RDD。默认情况下，只有8个并行任务来操作，但是可以传入一个可选的numTasks参数改变它。
      2. 需求：创建一个RDD，使用distinct()对其去重。*/
      def Test9():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val distinctRdd = sc.parallelize(List(1,2,1,5,2,9,6,1))
        //对RDD进行去重，不指定并行度
        val unionRDD = distinctRdd.distinct()
        println("不指定并行度"+unionRDD.collect().toBuffer)
    
        //添加并行度
        val unionRDD1 = distinctRdd.distinct(2)
        println("并行度为2"+unionRDD1.collect().toBuffer)
        sc.stop()
      }
      /*coalesce(numPartitions) 案例
        1. 作用：缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。
      2. 需求：创建一个4个分区的RDD，对其缩减分区*/
      def Test10():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(1 to 16,4)
        println( "分区数为："+rdd.partitions.size)//查看分区数
        //重新分区
        val coalesceRDD = rdd.coalesce(3)
        println("再次分区数为："+coalesceRDD.partitions.size)
        sc.stop()
      }
      /*sortBy(func,[ascending], [numTasks]) 案例
      1. 作用；使用func先对数据进行处理，按照处理后的数据比较结果排序，默认为正序。
      2. 需求：创建一个RDD，按照不同的规则进行排序*/
      def Test11():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(List(2,1,3,4))
        //按照自身大小排列
        val rdd1=rdd.sortBy(x => x)
        println(rdd1.collect().toBuffer)
        //按照与3余数的大小排序
        val rdd2=rdd.sortBy(x => x%3)
        println(rdd2.collect().toBuffer)
      }

**2 双Value类型交互**

/*1 union(otherDataset) 案例
        1. 作用：对源RDD和参数RDD求并集后返回一个新的RDD
      2. 需求：创建两个RDD，求并集*/
      def Num1():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd1 = sc.parallelize(1 to 5)
        val rdd2 = sc.parallelize(5 to 10)
        //计算两个RDD的并集
        val rdd3 = rdd1.union(rdd2)
        println(rdd3.collect().toBuffer)
        sc.stop()
      }
      /*2 subtract (otherDataset) 案例
        1. 作用：计算差的一种函数，去除两个RDD中相同的元素，不同的RDD将保留下来
      2. 需求：创建两个RDD，求第一个RDD与第二个RDD的差集*/
      def Num2():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(3 to 8)
        val rdd1 = sc.parallelize(1 to 5)
        //计算差集
        val rdd2=rdd.subtract(rdd1)
        println(rdd2.collect().toBuffer)
        sc.stop()
      }
      /*intersection(otherDataset) 案例
        1. 作用：对源RDD和参数RDD求交集后返回一个新的RDD
      2. 需求：创建两个RDD，求两个RDD的交集*/
      def Num3():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd=sc.parallelize(1 to 7)
        val rdd1=sc.parallelize(5 to 10)
        val rdd2= rdd1.intersection(rdd)//计算交集
        println(rdd2.collect().toBuffer)
        sc.stop()
      }
    
      /*zip(otherDataset)案例
        1. 作用：将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常。
      2. 需求：创建两个RDD，并将两个RDD组合到一起形成一个(k,v)RDD*/
      def Num4():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(Array(1,2,3),3)
        val rdd1=sc.parallelize(Array("a","b","c"),3)
        val rdd2=rdd.zip(rdd1)
        println(rdd2.collect().toBuffer)
        sc.stop()
      }

**3 Key-Value类型**

/*partitionBy案例
      1. 作用：对pairRDD进行分区操作，如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区， 否则会生成ShuffleRDD，即会产生shuffle过程。
      2. 需求：创建一个4个分区的RDD，对其重新分区*/
    
      def Ky1():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(Array((1,"aaa"),(2,"bbb"),(3,"ccc"),(4,"ddd")),4)
        var rdd2 = rdd.partitionBy(new org.apache.spark.HashPartitioner(2))
        println("rdd分区数为："+rdd.partitioner.size)
        println("rdd2分区数为："+rdd.partitioner.size)
        sc.stop()
      }
      /*groupByKey案例
      1. 作用：groupByKey也是对每个key进行操作，但只生成一个sequence。
      2. 需求：创建一个pairRDD，将相同key对应值聚合到一个sequence中，并计算相同key对应值的相加结果。*/
      def Ky2():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val words = Array("one", "two", "two", "three", "three", "three")
        val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))
        println(wordPairsRDD.collect().toBuffer)
        //将相同key对应值聚合到一个sequence中
        val group = wordPairsRDD.groupByKey()
        println(group.collect().toBuffer)
        //计算相同key对应值的相加结果
        val rdd1=group.map(t => (t._1, t._2.sum))
        println(rdd1.collect().toBuffer)
        sc.stop()
      }
      /*reduceByKey(func, [numTasks]) 案例
      1. 在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。
      2. 需求：创建一个pairRDD，计算相同key对应值的相加结果*/
      def Ky3():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(List(("female",1),("male",5),("female",5),("male",2)))
        //计算相同key对应值相加的结果
        val reduce= rdd.reduceByKey((x,y)=>x+y)
        println(reduce.collect().toBuffer)
        sc.stop()
      }
      /*sortByKey([ascending], [numTasks]) 案例
      1. 作用：在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
        2. 需求：创建一个pairRDD，按照key的正序和倒序进行排序*/
      def Ky4():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(Array((3,"aa"),(6,"cc"),(2,"bb"),(1,"dd")))
        //按照正序来排
        val rdd1=rdd.sortByKey(true)
        println("正序的结果为："+rdd1.collect().toBuffer)
        //倒序排列
        val rdd2=rdd.sortByKey(false)
        println("正序的结果为："+rdd2.collect().toBuffer)
        sc.stop()
      }
      /*mapValues案例
      1. 针对于(K,V)形式的类型只对V进行操作
        2. 需求：创建一个pairRDD，并将value添加字符串"|||"*/
      def Ky5():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd3 = sc.parallelize(Array((1,"a"),(1,"d"),(2,"b"),(3,"c")))
        println(rdd3.mapValues(_+"|||").collect().toBuffer)
        sc.stop()
      }
      /*join(otherDataset, [numTasks]) 案例
      1. 作用：在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD
        2. 需求：创建两个pairRDD，并将key相同的数据聚合到一个元组。*/
      def Ky6():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        //设置日志的级别
        sc.setLogLevel("WARN")
        val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
        val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))
        println(rdd.join(rdd1).collect().toBuffer)
        sc.stop()
      }
    
      //综合练习
      //需求：统计出每一个省份广告被点击次数的TOP3
      def Last_test():Unit={
        //1 创建配置文件对象
        val conf: SparkConf = new SparkConf().setAppName("rdd_create").setMaster("local[*]")
        //创建sparkcontext上下文
        val sc: SparkContext = new SparkContext(conf)
        System.setProperty("HADOOP_USER_NAME","hdfs")
        val line = sc.textFile("hdfs://clusterldg:8020/user/sheng/input/agent.log")
        //3.按照最小粒度聚合：((Province,AD),1)
        val provinceAdToOne = line.map { x =>
          val fields: Array[String] = x.split(" ")
          ((fields(1), fields(4)), 1)
        }
    
        //4.计算每个省中每个广告被点击的总数：((Province,AD),sum)
        val provinceAdToSum = provinceAdToOne.reduceByKey(_ + _)
    
        //5.将省份作为key，广告加点击数为value：(Province,(AD,sum))
        val provinceToAdSum = provinceAdToSum.map(x => (x._1._1, (x._1._2, x._2)))
    
        //6.将同一个省份的所有广告进行聚合(Province,List((AD1,sum1),(AD2,sum2)...))
        val provinceGroup = provinceToAdSum.groupByKey()
    
        //7.对同一个省份所有广告的集合进行排序并取前3条，排序规则为广告点击总数
        val provinceAdTop3 = provinceGroup.mapValues { x =>
          x.toList.sortWith((x, y) => x._2 > y._2).take(3)
        }
    
        //8.将数据拉取到Driver端并打印
        provinceAdTop3.collect().foreach(println)
    
        //9.关闭与spark的连接
        sc.stop()
      }