RDD的动作 Action

爱被打了一巴掌 2023-07-18 12:49 132阅读 0赞

一下整理了一些比较常用的Action  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU5OTM3Nw_size_16_color_FFFFFF_t_70]

package com.sheng
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    
    object Spark_Anction {
      def main(args: Array[String]): Unit = {
        Action10()
      }
      /*reduce(func)案例
        1. 作用：通过func函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据。
      2. 需求：创建一个RDD，将所有元素聚合得到结果。*/
      def Action1():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.makeRDD(1 to 10 ,2)
        //聚合RDD [int]所有元素
        val rdd2=rdd1.reduce(_+_)
        //创建一个RDD[string]并聚合
        val rdd3=sc.makeRDD(Array(("a",1),("a",3),("c",3),("d",5)))
        val rdd4=rdd3.reduce((x,y)=>(x._1 + y._1,x._2 + y._2))
        println(rdd2)
        println(rdd4)
        sc.stop()
      }
      /*count()案例
        1. 作用：返回RDD中元素的个数
      2. 需求：创建一个RDD，统计该RDD的条数*/
      def Action2():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)
        println(rdd.count())
        sc.stop()
      }
      /*first()案例
        1. 作用：返回RDD中的第一个元素
      2. 需求：创建一个RDD，返回该RDD中的第一个元素*/
      def Action3():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)
        println(rdd.first())//返回第一个元素
        sc.stop()
      }
     /* take(n)案例
        1. 作用：返回一个由RDD的前n个元素组成的数组
      2. 需求：创建一个RDD，统计该RDD的条数*/
      def Action4():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(2,5,4,6,8,3))
        println(rdd.take(3))
        sc.stop()
      }
      /*takeOrdered(n)案例
        1. 作用：返回该RDD排序后的前n个元素组成的数组
      2. 需求：创建一个RDD，统计该RDD的条数*/
      def Action5():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(2,5,4,6,8,3))
        println(rdd.takeOrdered(3))
        sc.stop()
      }
      /*aggregate案例
      1. 参数：(zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)
      2. 作用：aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合，然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致。
      3. 需求：创建一个RDD，将所有元素相加得到结果*/
      def Action6():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 rdd1 = sc.makeRDD(1 to 10,2)
        println(rdd1.aggregate(0)(_+_,_+_))
        sc.stop()
      }
      /*fold(num)(func)案例
        1. 作用：折叠操作，aggregate的简化操作，seqop和combop一样。
      2. 需求：创建一个RDD，将所有元素相加得到结果*/
      def Action7():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 rdd1 = sc.makeRDD(1 to 10,2)
        println(rdd1.fold(1)(_+_))
        sc.stop()
      }
     /* saveAsTextFile(path)
      作用：将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本
    */
      def Action8():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)
        rdd.saveAsTextFile("hdfs://clusterldg:8020/user/sheng/input_spark")
        println("保存完成")
        sc.stop()
      }
     /* countByKey()案例
        1. 作用：针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。
      2. 需求：创建一个PairRDD，统计每种key的个数*/
      def Action9():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((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)
        val rdd1= sc.parallelize(Array(("xiaomi",1),("xiaomi",2),("hongmi",1)))
        println( rdd.countByKey)
        println(rdd1.countByKey())
        sc.stop()
      }
      /*foreach(func)案例
        1. 作用：在数据集的每一个元素上，运行函数func进行更新。
      2. 需求：创建一个RDD，对每个元素进行打印*/
      def Action10():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 rdd = sc.makeRDD(1 to 5,2)
        rdd.foreach(println(_))
    	sc.stop()
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzU5OTM3Nw_size_16_color_FFFFFF_t_70]: /images/20230528/6d61ef5f37dc42169c93f6b919eaa623.png