Spark核心编程与项目案例详解（一）

灰太狼 2022-12-24 09:53 189阅读 0赞

### 目录 ###

*   *  一、Spark结构原理
     *  二、创建RDD的几种方式
     *   *  （1）Parallelize集合
         *  （2）本地或者HDFS文件
     *  三、Transformation算子
     *  四、Action算子
     *  五、RDD持久化详解
     *   *  （1）持久化原理
         *  （2）cache和persist的区别
         *  （3）持久化策略
         *  （4）如何选择RDD持久化策略
         *  （5）测试RDD持久化

## 一、Spark结构原理 ##

*  1.Driver进程启动之后，会做一些初始化的操作。在这个操作中，就会发送请求到Master进行spark应用程序的注册；目的是告诉master主节点，有一个新的spark应用程序要运行。
 *  2.Master在接受到spark应用程序的注册申请之后，会发送请求给Worker进行资源的调度和分配，也即是Executor的分配
 *  3.Worker接受到Master的请求之后，会为spark应用程序启动Executor。
 *  4.Executor启动之后，会向Driver进行反向注册。这样driver就知道有哪些Executor正在为它服务了
 *  5.Driver注册了一些Executor之后，就可以开始正式执行我们的spark应用程序了它首先要做的，就是读取HDFS的数据来创建初始RDD。
 *  6.HDFS文件内容被读取到多个worker节点上，形成内存中的分布式数据集，也即是RDD partition
 *  7.Driver会根据我们对RDD定义的操作，提交一大堆Task到Executor。

## 二、创建RDD的几种方式 ##

进行Spark核心编程时，首先要做的第一件事，就是创建一个初始的RDD。该RDD中，通常就代表和包含了Spark应用程序的输入源数据。然后在创建了初始的RDD之后，才可以通过Spark Core提供的transformation算子，对该RDD进行转换，来获取其他的RDD。

Spark Core提供了三种创建RDD的方式，包括：

*  使用程序中的集合创建RDD
 *  使用本地文件创建RDD
 *  使用HDFS文件创建RDD

个人经验认为：

*  1、使用程序中的集合创建RDD，主要用于进行测试，可以在实际部署到集群运行之前，自己使用集合构造测试数据，来测试后面的spark应用的流程。
 *  2、使用本地文件创建RDD，主要用于临时性地处理一些存储了大量数据的文件。
 *  3、使用HDFS文件创建RDD，应该是最常用的生产环境处理方式，主要可以针对HDFS上存储的大数据，进行离线批处理操作。

### （1）Parallelize集合 ###

如果要通过并行化集合来创建RDD，需要针对程序中的集合，调用SparkContext的parallelize()方法。 Spark会将集合中的数据拷贝到集群上去，形成一个分布式的数据集合，也就是一个RDD。相当于是，集合中的部分数据会到一个节点上，而另一部分数据会到其他节点上。然后就可以用并行的方式来操作这个分布式数据集合，即RDD。

// 案例： 1到10累加求和
    val arr= Array(1,2, 3, 4, 5,6,7,8, 9, 10)
    val rdd = sc.parallelize(arr)
    val sum = rdd.reduce(_ + _)

调用parallelize()时，有一个重要的参数可以指定，就是要将集合切分成多少个partition。Spark会 为每一个parition运行一个task来进行处理。Spark官方的建议是，为集群中的每个CPU创建2~4个partition。Spark默认 会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时，传入第二个参数，来设置RDD的partition数量。比如pallelize(arr,10)

### （2）本地或者HDFS文件 ###

Spark是支持使用任何Hadoop支持的存储系统上的文件创建RDD的，比如说HDFS、Cassandra、 HBase以及 本地文件。通过调用SparkContext的\*\*textFile()\*\*方法，可以针对本地文件或HDFS文件创建RDD。  
有几个事项是需要注意的：

*  1、如果是针对本地文件的话，如果是在windows上本地测试，windows上有一份文件即可；如果是在spark集群上针对linux本地文件，那么需要将文件拷贝到所有worker节点上。
 *  2、Spark的textFile()方法支持针对目录、压缩文件以及通配符进行RDD创建。
 *  3、Spark默认会为hdfs文件的每一个block创建一个partition，但是也可以通过textFile()的第二个参数手动设置分区数量，只能比block数量多，不能比block数量少。

//案例：文件字数统计
    val rdd = sc.textFile("data.txt")
    val wordCount = rdd.map(line => line.length).reduce(_ + _)

## 三、Transformation算子 ##

Spark支持两种RDD操作： **transformation和action**

transformation操作会针对已有的RDD创建一个新的RDD；而action则主要是对RDD进行最后的操作，比如遍历、reduce、 保存到文件等，并可以返回结果给Driver程序。

例如，map就是一种transformnation操作，它用于将已有RDD的每个元素传入一个自定义的函数，并获取一个新的元素，然后将所有的新元素组成一个新的RDD。而reduce就是一种action操作，它用于对RDD中的所有元素进行聚合操作，并获取一个最终的结果，然后返回给Driver程序。

transformation的特点就是lazy特性。lazy特性 指的是，如果一个spark应用中只定义了transformation操作，那么即使你执行该应用，这些操作也不会执行。也就是说，transformation是 不会触发spark程序的执行的，它们只是记录了对RDD所做的操作，但是不会自发的执行。只有当transformation之后，接着执行了一个action操作，那么所有的transformation才会执行。Spark通过 这种lazy特性，来进行底层的spark应用执行的优化，避免产生过多中间结果。

action操作执行，会触发一个spark job的运行，从而触发这个action之前所有的transformation的执行。这是action的特性。

常见的Transformation算子：

<table> 
 <thead> 
  <tr> 
   <th>Transformation</th> 
   <th>Feature</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>map</td> 
   <td>将RDD中的每个元素传入自定义函数，获取一个新的元素，然后用新的元素组成新的RDD</td> 
  </tr> 
  <tr> 
   <td>filter</td> 
   <td>对RDD中每个元素进行判断，如果返回true则保留，返回false则剔除。</td> 
  </tr> 
  <tr> 
   <td>flatMap</td> 
   <td>与map类似，但是对每个元素都可以返回一个或多个新元素。</td> 
  </tr> 
  <tr> 
   <td>gropuByKey</td> 
   <td>根据key进行分组，每个key对应一个Iterable</td> 
  </tr> 
  <tr> 
   <td>reduceByKey</td> 
   <td>对每个key对应的value进行reduce操作。</td> 
  </tr> 
  <tr> 
   <td>sortByKey</td> 
   <td>对每个key对应的value进行排序操作。</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>对两个包含&lt;key，value&gt;对的RDD进行join操作，每个key join上的pair，都会传入自定义函数进行处理。</td> 
  </tr> 
  <tr> 
   <td>cogroup</td> 
   <td>同join，但是是每个key对应的Iterable都会传入自定义函数进行处理。</td> 
  </tr> 
 </tbody> 
</table>

**使用Java语言对每一种Transformation算子举例讲解：**

package com.kfk.spark.core;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.FlatMapFunction;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.Arrays;
    import java.util.Iterator;
    import java.util.List;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/25 * @time : 7:10 下午 */
    public class TransformationJava { 
    
        public static JavaSparkContext getsc(){ 
            SparkConf sparkConf = new SparkConf().setAppName("TransformationJava").setMaster("local");
            return new JavaSparkContext(sparkConf);
        }
    
        public static void main(String[] args) { 
    
             map();
             filter();
             flatmap();
             groupByKey();
             reduceByKey();
             sortByKey();
             join();
             cogroup();
    
        }
    
        /** * 数据集一：(2,"lili") cogroup() -> <2,<"lili",(90,95,99)>> * 数据集二：(2,90)(2,95)(2,99) */
        private static void cogroup() { 
            List stuList = Arrays.asList(new Tuple2<Integer,String>(1,"alex"),
                    new Tuple2<Integer,String>(2,"lili"),
                    new Tuple2<Integer,String>(3,"cherry"),
                    new Tuple2<Integer,String>(4,"jack"),
                    new Tuple2<Integer,String>(5,"jone"),
                    new Tuple2<Integer,String>(6,"lucy"),
                    new Tuple2<Integer,String>(7,"aliy"));
    
            List scoreList = Arrays.asList(new Tuple2<Integer,Integer>(1,90),
                    new Tuple2<Integer,Integer>(2,79),
                    new Tuple2<Integer,Integer>(2,95),
                    new Tuple2<Integer,Integer>(2,99),
                    new Tuple2<Integer,Integer>(3,87),
                    new Tuple2<Integer,Integer>(3,88),
                    new Tuple2<Integer,Integer>(3,89),
                    new Tuple2<Integer,Integer>(4,98),
                    new Tuple2<Integer,Integer>(5,89),
                    new Tuple2<Integer,Integer>(6,93),
                    new Tuple2<Integer,Integer>(7,96));
    
            JavaSparkContext sc = getsc();
    
            JavaPairRDD javaPairStuRdd = sc.parallelizePairs(stuList);
            JavaPairRDD javaPairScoreRdd = sc.parallelizePairs(scoreList);
    
            JavaPairRDD<Integer,Tuple2<Iterable<Integer>,Iterable<Integer>>> cogroupValues = javaPairStuRdd.cogroup(javaPairScoreRdd);
            cogroupValues.foreach(new VoidFunction<Tuple2<Integer, Tuple2<Iterable<Integer>, Iterable<Integer>>>>() { 
                public void call(Tuple2<Integer, Tuple2<Iterable<Integer>, Iterable<Integer>>> integerTuple2Tuple2) throws Exception { 
                    System.out.println(integerTuple2Tuple2._1);
                    System.out.println(integerTuple2Tuple2._2._1 + " : " + integerTuple2Tuple2._2._2);
                }
            });
        }
    
        /** * 数据集一：(1,"alex") join() -> <1,<"alex",90>> * 数据集二：(1,90) */
        private static void join() { 
    
            List stuList = Arrays.asList(new Tuple2<Integer,String>(1,"alex"),
                    new Tuple2<Integer,String>(2,"lili"),
                    new Tuple2<Integer,String>(3,"cherry"),
                    new Tuple2<Integer,String>(4,"jack"),
                    new Tuple2<Integer,String>(5,"jone"),
                    new Tuple2<Integer,String>(6,"lucy"),
                    new Tuple2<Integer,String>(7,"aliy"));
    
            List scoreList = Arrays.asList(new Tuple2<Integer,Integer>(1,90),
                    new Tuple2<Integer,Integer>(2,95),
                    new Tuple2<Integer,Integer>(3,87),
                    new Tuple2<Integer,Integer>(4,98),
                    new Tuple2<Integer,Integer>(5,89),
                    new Tuple2<Integer,Integer>(6,93),
                    new Tuple2<Integer,Integer>(7,96));
    
            JavaSparkContext sc = getsc();
    
            JavaPairRDD javaPairStuRdd = sc.parallelizePairs(stuList);
            JavaPairRDD javaPairScoreRdd = sc.parallelizePairs(scoreList);
    
            JavaPairRDD<Integer,Tuple2<String,Integer>> joinValue = javaPairStuRdd.join(javaPairScoreRdd);
    
            joinValue.foreach(new VoidFunction<Tuple2<Integer, Tuple2<String, Integer>>>() { 
                public void call(Tuple2<Integer, Tuple2<String, Integer>> integerTuple2Tuple2) throws Exception { 
                    System.out.println(integerTuple2Tuple2._1);
                    System.out.println(integerTuple2Tuple2._2._1 + " : " + integerTuple2Tuple2._2._2);
                }
            });
        }
    
        /** * <90,alex> * <95,lili> sortByKey() -> <87,cherry> <90,alex> <95,lili> * <87,cherry> */
        private static void sortByKey() { 
    
            List list = Arrays.asList(new Tuple2<Integer,String>(90,"alex"),
                    new Tuple2<Integer,String>(95,"lili"),
                    new Tuple2<Integer,String>(87,"cherry"),
                    new Tuple2<Integer,String>(98,"jack"),
                    new Tuple2<Integer,String>(89,"jone"),
                    new Tuple2<Integer,String>(93,"lucy"),
                    new Tuple2<Integer,String>(96,"aliy"));
    
            JavaPairRDD<Integer, String> javaPairRdd = getsc().parallelizePairs(list);
            JavaPairRDD<Integer,String> sortByKeyValues = javaPairRdd.sortByKey(true);
    
            sortByKeyValues.foreach(new VoidFunction<Tuple2<Integer, String>>() { 
                public void call(Tuple2<Integer, String> integerStringTuple2) throws Exception { 
                    System.out.println(integerStringTuple2._1 + " : " + integerStringTuple2._2);
                }
            });
        }
    
        /** * <class_1,(90,87,98,96)> reduceByKey() -> <class_1,(90+87+98+96)> * <class_2,(95,89,93)> reduceByKey() -> <class_2,(95+89+93)> */
        private static void reduceByKey() { 
    
            List list = Arrays.asList(new Tuple2<String,Integer>("class_1",90),
                    new Tuple2<String,Integer>("class_2",95),
                    new Tuple2<String,Integer>("class_1",87),
                    new Tuple2<String,Integer>("class_1",98),
                    new Tuple2<String,Integer>("class_2",89),
                    new Tuple2<String,Integer>("class_2",93),
                    new Tuple2<String,Integer>("class_1",96));
    
            final JavaPairRDD<String, Integer> javaPairRdd = getsc().parallelizePairs(list);
    
            JavaPairRDD<String,Integer> reduceByKeyValues = javaPairRdd.reduceByKey(new Function2<Integer,Integer,Integer>() { 
                public Integer call(Integer v1, Integer v2) throws Exception { 
                    return v1+v2;
                }
            });
    
            reduceByKeyValues.foreach(new VoidFunction<Tuple2<String, Integer>>() { 
                public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception { 
                    System.out.println(stringIntegerTuple2._1 + " : " + stringIntegerTuple2._2);
                }
            });
        }
    
        /** * class_1 90 groupByKey() -> <class_1,(90,87,98,96)> <class_2,(95,89,93)> * class_2 95 * class_1 87 * class_1 98 * class_2 89 * class_2 93 * class_1 96 */
        private static void groupByKey() { 
    
            List list = Arrays.asList(new Tuple2<String,Integer>("class_1",90),
                    new Tuple2<String,Integer>("class_2",95),
                    new Tuple2<String,Integer>("class_1",87),
                    new Tuple2<String,Integer>("class_1",98),
                    new Tuple2<String,Integer>("class_2",89),
                    new Tuple2<String,Integer>("class_2",93),
                    new Tuple2<String,Integer>("class_1",96));
    
            JavaPairRDD<String,Integer> javaPairRdd = getsc().parallelizePairs(list);
    
            JavaPairRDD<String,Iterable<Integer>> groupByKeyValue = javaPairRdd.groupByKey();
    
            groupByKeyValue.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() { 
                public void call(Tuple2<String,Iterable<Integer>> stringIteratorTuple2) throws Exception { 
                    System.out.println(stringIteratorTuple2._1);
    
                    Iterator<Integer> iterator = stringIteratorTuple2._2.iterator();
                    while (iterator.hasNext()){ 
                        System.out.println(iterator.next());
                    }
                }
            });
        }
    
        /** * hbase hadoop hive * java python flatmap() -> hbase hadoop hive java python java python * java python */
        private static void flatmap() { 
    
            List<String> list = Arrays.asList("hbase hadoop hive","java python","java python");
    
            JavaRDD<String> javaRdd = getsc().parallelize(list);
    
            JavaRDD<String> flatMapValue = javaRdd.flatMap(new FlatMapFunction<String,String>() { 
                public Iterator<String> call(String line) throws Exception { 
                    return Arrays.asList(line.split(" ")).iterator();
                }
            });
    
            flatMapValue.foreach(new VoidFunction<String>() { 
                public void call(String value) throws Exception { 
                    System.out.println(value);
                }
            });
        }
    
        /** * 1,2,3,4,5,6,7,8,9,10 filter() -> 2,4,6,8,10 */
        private static void filter() { 
    
            List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
            JavaRDD<Integer> javaRdd = getsc().parallelize(list);
    
            // 取偶数
            JavaRDD<Integer> filterValue = javaRdd.filter(new Function<Integer,Boolean>() { 
                public Boolean call(Integer value) throws Exception { 
                    return value % 2 == 0;
                }
            });
    
            filterValue.foreach(new VoidFunction<Integer>() { 
                public void call(Integer o) throws Exception { 
                    System.out.println(o);
                }
            });
        }
    
        /** * 1,2,3,4,5 map() -> 10,20,30,40,50 */
        public static void map(){ 
    
            List<Integer> list = Arrays.asList(1,2,3,4,5);
            JavaRDD<Integer> javaRdd = getsc().parallelize(list);
    
            JavaRDD<Integer> mapValue = javaRdd.map(new Function<Integer,Integer>() { 
                public Integer call(Integer value) throws Exception { 
                    return value * 10;
                }
            });
    
            mapValue.foreach(new VoidFunction<Integer>() { 
                public void call(Integer o) throws Exception { 
                    System.out.println(o);
                }
            });
        }
    }

**使用Scala语言对每一种Transformation算子举例讲解：**

package com.kfk.spark.core
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/25 * @time : 7:19 下午 */
    object TransformationScala { 
    
        def getsc():SparkContext ={ 
            val sparkConf = new SparkConf().setAppName("TransformationScala").setMaster("local")
            return new SparkContext(sparkConf)
        }
    
        def main(args: Array[String]): Unit = { 
            map()
            filter()
            flatmap()
            groupByKey()
            reduceByKey()
            sortByKey()
            join()
            cogroup()
        }
    
        /** * 1,2,3,4,5 map() -> 10,20,30,40,50 */
        def map(): Unit ={ 
    
            val list = Array(1,2,3,4,5)
    
            val rdd = getsc().parallelize(list)
    
            val mapValue = rdd.map(x => x * 10)
    
            mapValue.foreach(x => System.out.println(x))
        }
    
        /** * 1,2,3,4,5,6,7,8,9,10 filter() -> 2,4,6,8,10 */
        def filter(): Unit ={ 
    
            val list = Array(1,2,3,4,5,6,7,8,9,10)
    
            val rdd = getsc().parallelize(list)
    
            val filterValue = rdd.filter(x => x % 2 == 0)
    
            filterValue.foreach(x => System.out.println(x))
        }
    
        /** * hbase hadoop hive * java python flatmap() -> hbase hadoop hive java python java python * java python */
        def flatmap(): Unit = { 
    
            val list = Array("hbase hadoop hive", "java python", "storm spark")
    
            val rdd = getsc().parallelize(list)
    
            val flatMapValue = rdd.flatMap(x => x.split(" "))
    
            flatMapValue.foreach(x => System.out.println(x))
        }
    
        /** * class_1 90 groupByKey() -> <class_1,(90,87,98,96)> <class_2,(95,89,93)> * class_2 95 * class_1 87 * class_1 98 * class_2 89 * class_2 93 * class_1 96 */
        def groupByKey(): Unit = { 
    
            val list = Array(
                Tuple2("class_1", 90),
                Tuple2("class_2", 95),
                Tuple2("class_1", 87),
                Tuple2("class_1", 98),
                Tuple2("class_2", 89),
                Tuple2("class_2", 93),
                Tuple2("class_1", 96))
    
            val rdd = getsc().parallelize(list)
    
            val groupByKeyValue = rdd.groupByKey()
    
            groupByKeyValue.foreach(x => { 
                System.out.println(x._1)
                x._2.foreach(y => System.out.println(y))
            })
        }
    
        /** * <class_1,(90,87,98,96)> reduceByKey() -> <class_1,(90+87+98+96)> * <class_2,(95,89,93)> reduceByKey() -> <class_2,(95+89+93)> */
        def reduceByKey(): Unit = { 
    
            val list = Array(
                Tuple2("class_1", 90),
                Tuple2("class_2", 95),
                Tuple2("class_1", 87),
                Tuple2("class_1", 98),
                Tuple2("class_2", 89),
                Tuple2("class_2", 93),
                Tuple2("class_1", 96))
    
            val rdd = getsc().parallelize(list)
    
            val reduceByKeyValues = rdd.reduceByKey((x,y) => x+y)
    
            reduceByKeyValues.foreach(x => { 
                System.out.println(x._1 + " : " + x._2)
            })
        }
    
        /** * <90,alex> * <95,lili> -> <87,cherry> <90,alex> <95,lili> * <87,cherry> */
        def sortByKey(): Unit ={ 
    
            val list = Array(Tuple2(90, "alex"),
                Tuple2(95, "lili"),
                Tuple2(87, "cherry"),
                Tuple2(98, "jack"),
                Tuple2(89, "jone"),
                Tuple2(93, "lucy"),
                Tuple2(96, "aliy")
            )
    
            val rdd = getsc().parallelize(list)
    
            val sortByKeyValues = rdd.sortByKey(true)
            sortByKeyValues.foreach(x => { 
                System.out.println(x._1 + " : " + x._2)
            })
        }
    
        /** * 数据集一：(1,"alex") join() -> <1,<"alex",90>> * 数据集二：(1,90) */
        def join(): Unit ={ 
    
            val stuList = Array(Tuple2(1, "alex"),
                Tuple2(2, "lili"),
                Tuple2(3, "cherry"),
                Tuple2(4, "jack"),
                Tuple2(5, "jone"),
                Tuple2(6, "lucy"),
                Tuple2(7, "aliy"))
    
            val scoreList = Array(Tuple2(1, 90),
                Tuple2(2, 95),
                Tuple2(3, 87),
                Tuple2(4, 98),
                Tuple2(5, 89),
                Tuple2(6, 93),
                Tuple2(7, 96))
    
            val sc = getsc()
    
            val stuRdd = sc.parallelize(stuList)
    
            val scoreRdd = sc.parallelize(scoreList)
    
            val joinValue = stuRdd.join(scoreRdd)
            joinValue.foreach(x => { 
                System.out.println(x._1 + " > " + x._2._1 + " : " + x._2._2)
            })
        }
    
        /** * 数据集一：(2,"lili") cogroup() -> <2,<"lili",(90,95,99)>> * 数据集二：(2,90)(2,95)(2,99) */
        def cogroup(): Unit ={ 
    
            val stuList = Array(Tuple2(1, "alex"),
                Tuple2(2, "lili"),
                Tuple2(3, "cherry"),
                Tuple2(4, "jack"),
                Tuple2(5, "jone"),
                Tuple2(6, "lucy"),
                Tuple2(7, "aliy"))
    
            val scoreList = Array(Tuple2(1, 90),
                Tuple2(2, 95),
                Tuple2(2, 95),
                Tuple2(2, 99),
                Tuple2(3, 87),
                Tuple2(3, 88),
                Tuple2(3, 89),
                Tuple2(4, 98),
                Tuple2(5, 89),
                Tuple2(6, 93),
                Tuple2(7, 96))
    
            val sc = getsc()
    
            val stuRdd = sc.parallelize(stuList)
    
            val scoreRdd = sc.parallelize(scoreList)
    
            val cogroupValues = stuRdd.cogroup(scoreRdd)
            cogroupValues.foreach(x => { 
                System.out.println(x._1 + " > " + x._2._1.toList + " : " + x._2._2.toList)
            })
        }
    }

## 四、Action算子 ##

常见Action算子：

<table> 
 <thead> 
  <tr> 
   <th>Action</th> 
   <th>Feature</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>reduce</td> 
   <td>将RDD中的所有元素进行聚合操作。第一个和第二个元素聚合，值与第三个元素聚合，值与第四个元素聚合，以此类推。</td> 
  </tr> 
  <tr> 
   <td>collect</td> 
   <td>将RDD中所有元素获取到本地客户端。</td> 
  </tr> 
  <tr> 
   <td>count</td> 
   <td>获取RDD元素总数。</td> 
  </tr> 
  <tr> 
   <td>take(n)</td> 
   <td>获取RDD中前n个元素。</td> 
  </tr> 
  <tr> 
   <td>saveAsTextFile</td> 
   <td>将RDD元素保存到文件中，对每个元素调用toString方法。</td> 
  </tr> 
  <tr> 
   <td>countByKey</td> 
   <td>对每个key对应的值进行count计数。</td> 
  </tr> 
  <tr> 
   <td>foreach</td> 
   <td>遍历RDD中的每个元素。</td> 
  </tr> 
 </tbody> 
</table>

**使用Java语言对每一种Action算子举例讲解：**

**使用Scala语言对每一种Action算子举例讲解：**

package com.kfk.spark.core
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/26 * @time : 10:42 下午 */
    object ActionScala { 
    
        def getsc():SparkContext ={ 
            val sparkConf = new SparkConf().setAppName("TransformationScala").setMaster("local")
            return new SparkContext(sparkConf)
        }
    
        def main(args: Array[String]): Unit = { 
    
            reduce()
            collect()
            count()
            take()
            save()
            countByKey()
        }
    
        /** * 1,2,3,4,5 reduce() -> 15 */
        def reduce(): Unit = { 
    
            val list = Array(1,2,3,4,5)
    
            val rdd = getsc().parallelize(list)
    
            val reduceValues = rdd.reduce((x,y) => x + y)
    
            System.out.println(reduceValues)
        }
    
        /** * 1,2,3,4,5 collect() -> [1,2,3,4,5] */
        def collect(): Unit ={ 
    
            val list = Array(1,2,3,4,5)
    
            val rdd = getsc().parallelize(list)
    
            val collectValue = rdd.collect()
    
            for (value <- collectValue){ 
                System.out.println(value)
            }
        }
    
        /** * 1,2,3,4,5 count() -> 5 */
        def count(): Unit ={ 
    
            val list = Array(1,2,3,4,5)
    
            val rdd = getsc().parallelize(list)
    
            System.out.println(rdd.count())
        }
    
        /** * 1,2,3,4,5 take(3) -> [1,2,3] */
        def take(): Unit ={ 
    
            val list = Array(1,2,3,4,5)
    
            val rdd = getsc().parallelize(list)
    
            val takeValues = rdd.take(3)
    
            for (value <- takeValues){ 
                System.out.println(value)
            }
        }
    
        /** * saveAsTextFile() */
        def save(): Unit ={ 
    
            val list = Array(1,2,3,4,5)
    
            val rdd = getsc().parallelize(list)
    
            rdd.saveAsTextFile("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/scalaRdd")
        }
    
        /** * <"class_1","alex"> * <"class_2","jone"> * <"class_1","lucy"> <class_1,4> * <"class_1","lili"> countByKey() -> * <"class_2","ben"> <class_2,3> * <"class_2","jack"> * <"class_1","cherry"> */
        def countByKey(): Unit ={ 
    
            val list = Array(
                Tuple2("class_1", "alex"),
                Tuple2("class_2", "jack"),
                Tuple2("class_1", "jone"),
                Tuple2("class_1", "lili"),
                Tuple2("class_2", "ben"),
                Tuple2("class_2", "lucy"),
                Tuple2("class_1", "cherry"))
    
            val rdd = getsc().parallelize(list)
    
            val countByKeyValues = rdd.countByKey()
    
            System.out.println(countByKeyValues)
        }
    }

## 五、RDD持久化详解 ##

### （1）持久化原理 ###

Spark非常重要的一个功能特性就是可以将RDD持久化在内存中。当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存缓存的partition。这样的话，对于针对一个RDD反复执行多个操作的场景，就只要对RDD计算一次即可，后面直接使用该RDD，而不需要反复计算多次该RDD。

巧妙使用RDD持久化，甚至在某些场景下，可以将spark应用程序的性能提升10倍。对于迭代式算法和快速交互式应用来说，RDD持久化，是非常重要的。

要持久化一个RDD，只要调用其cache()或者persist()方法即可。在该RDD第一次被计算出来时，就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的，如果持久化的RDD的任何parition丢失了，那么Spark会自动通过其源RDD，使用transformation操作重新计算该partition。

**不使用持久化**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]  
**使用持久化**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

### （2）cache和persist的区别 ###

基于Spark 1.6.1 的源码，可以看到

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
    def cache(): this.type = persist()

说明是cache()调用了persist(), 想要知道二者的不同还需要看一下persist函数：

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
    def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

可以看到persist()内部调用了persist(StorageLevel.MEMORY\_ONLY)，继续深入：

/** * Set this RDD's storage level to persist its values across operations after the first time * it is computed. This can only be used to assign a new storage level if the RDD does not * have a storage level set yet.. */
    def persist(newLevel: StorageLevel): this.type = { 
      // TODO: Handle changes of StorageLevel
      if (storageLevel != StorageLevel.NONE && newLevel != storageLevel) { 
        throw new UnsupportedOperationException(
          "Cannot change storage level of an RDD after it was already assigned a level")
      }
      sc.persistRDD(this)
      // Register the RDD with the ContextCleaner for automatic GC-based cleanup
      sc.cleaner.foreach(_.registerRDDForCleanup(this))
      storageLevel = newLevel
      this
    }

可以看出来persist有一个 StorageLevel 类型的参数，这个表示的是RDD的缓存级别。

至此便可得出cache和persist的区别了：**cache只有一个默认的缓存级别MEMORY\_ONLY ，而persist可以根据情况设置其它的缓存级别。**

### （3）持久化策略 ###

RDD持久化是可以手动选择不同的策略的。比如可以将RDD持久化在内存中、持久化到磁盘上、使用序列化的方式持久化，多持久化的数据进行多路复用。只要在调用persist()时传入对应的StorogeLevel即可。

<table> 
 <thead> 
  <tr> 
   <th>持久化级别</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>MEMORY_ONLY</td> 
   <td>以非序列化的Java对象的方式持久化在JVM内存中。如果内存无法完全存储RDD所有的partition，那么那些没有持久化的partition就会在下次需要使用它的时候，重新被计算。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK</td> 
   <td>同上，但是当某些partition无法存储在内存中时，会持久化到磁盘中。下次需要使用这些partition时，需要从磁盘上读取。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER</td> 
   <td>同MEMORY ONLY，但是会使用Java序列化方式，将Java对象序列化后进行持久化。可以减少内存开销，但是需要进行反序列化，因此会加大CPU开销。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DSK_SER</td> 
   <td>同MEMORY_ AND DSK，但是使用序列化方式持久化Java对象。</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY</td> 
   <td>使用非序列化Java对象的方式持久化，完全存储到磁盘上。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_2 MEMORY_ AND_DISK_2</td> 
   <td>如果是尾部加了2的持久化级别，表示会将持久化数据复用一份，保存到其他节点。从而在数据丢失时，不需要再次计算，只需要使用备份数据即可。</td> 
  </tr> 
 </tbody> 
</table>

顺便看一下RDD都有哪些缓存级别，查看 StorageLevel 类的源码：

object StorageLevel { 
      val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(false, false, true, false)
      ......
    }

可以看到这里列出了12种缓存级别，但这些有什么区别呢？可以看到每个缓存级别后面都跟了一个StorageLevel的构造函数，里面包含了4个或5个参数，如下

val MEMORY_ONLY = new StorageLevel(false, true, false, true)

查看其构造函数

class StorageLevel private(
        private var _useDisk: Boolean,
        private var _useMemory: Boolean,
        private var _useOffHeap: Boolean,
        private var _deserialized: Boolean,
        private var _replication: Int = 1)
      extends Externalizable { 
      ......
      def useDisk: Boolean = _useDisk
      def useMemory: Boolean = _useMemory
      def useOffHeap: Boolean = _useOffHeap
      def deserialized: Boolean = _deserialized
      def replication: Int = _replication
      ......
    }

可以看到StorageLevel类的主构造器包含了5个参数：

*  useDisk：使用硬盘（外存）
 *  useMemory：使用内存
 *  useOffHeap：使用堆外内存，这是Java虚拟机里面的概念，堆外内存意味着把内存对象分配在Java虚拟机的堆以外的内存，这些内存直接受操作系统管理（而不是虚拟机）。这样做的结果就是能保持一个较小的堆，以减少垃圾收集对应用的影响。
 *  deserialized：反序列化，其逆过程序列化（Serialization）是java提供的一种机制，将对象表示成一连串的字节；而反序列化就表示将字节恢复为对象的过程。序列化是对象永久化的一种机制，可以将对象及其属性保存起来，并能在反序列化后直接恢复这个对象
 *  replication：备份数（在多个节点上备份）

理解了这5个参数，StorageLevel 的12种缓存级别就不难理解了。

val MEMORY\_AND\_DISK\_SER\_2 = new StorageLevel(true, true, false, false, 2) 就表示使用这种缓存级别的RDD将存储在硬盘以及内存中，使用序列化（在硬盘中），并且在多个节点上备份2份（正常的RDD只有一份）

另外还注意到有一种特殊的缓存级别

val OFF_HEAP = new StorageLevel(false, false, true, false)

使用了堆外内存，StorageLevel 类的源码中有一段代码可以看出这个的特殊性，它不能和其它几个参数共存。

if (useOffHeap) { 
      require(!useDisk, "Off-heap storage level does not support using disk")
      require(!useMemory, "Off-heap storage level does not support using heap memory")
      require(!deserialized, "Off-heap storage level does not support deserialized storage")
      require(replication == 1, "Off-heap storage level does not support multiple replication")
    }

### （4）如何选择RDD持久化策略 ###

Spark提供的多种持久化级别，主要是为了在CPU和内存消耗之间进行取舍。下面是一些通用的持久化级别的选择建议：

*  1、优先使用MEMORY\_ONLY，如果可以缓存所有数据的话，那么就使用这种策略。因为纯内存速度最快，而且没有序列化，不需要消耗CPU进行反序列化操作。
 *  2、如果MEMORY\_ ONLY策略，无法存储的下所有数据的话，那么使用MEMORY\_ONLY\_SER，将数据进行序列化进行存储，纯内存操作还是非常快，只是要消耗CPU进行反序列化。
 *  3、如果需要进行快速的失败恢复，那么就选择带后缀为\_2的策略，进行数据的备份，这样在失败时，就不需要重新计算了。
 *  4、能不使用DISK相关的策略，就不用使用，有的时候，从磁盘读取数据，还不如重新计算一次。

### （5）测试RDD持久化 ###

下面我将用代码来验证一下RDD持久化的优势

package com.kfk.spark.core;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    
    import javax.swing.*;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/27 * @time : 2:35 下午 */
    public class persistJava { 
    
        public static JavaSparkContext getsc(){ 
            SparkConf sparkConf = new SparkConf().setAppName("ActionJava").setMaster("local");
            return new JavaSparkContext(sparkConf);
        }
        public static void main(String[] args) { 
            JavaRDD lines = getsc().textFile("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/2015082818").cache();
    
            // 第一次开始时间
            long begin = System.currentTimeMillis();
            System.out.println("行数:"+lines.count());
            // 第一次总时间
            System.out.println("第一次总时间:"+(System.currentTimeMillis() - begin));
    
            // 第二次开始时间
            long begin1 = System.currentTimeMillis();
            System.out.println("行数:"+lines.count());
            // 第二次总时间
            System.out.println("第二次总时间:"+(System.currentTimeMillis() - begin1));
        }
    }

不使用rdd持久化：

行数:64972
    第一次总时间:1085

行数:64972
    第二次总时间:191

使用rdd持久化：

行数:64972
    第一次总时间:986

行数:64972
    第二次总时间:23

测试的话尽量采用数据量大一些，不然测试结果会看不出效果，根据上面的测试结果，我们可以看出使用rdd持久化和不使用rdd持久化差距很明显。

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/39eb586cac0742c28998bbfc6708a245.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201127184513241.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center