devbox
凡人多烦事01
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

spark中RDD编程（java）_javardd groupby

作者：凡人多烦事01 | 2024-04-22 03:51:46

赞

踩

javardd groupby

目录

一、RDD的概念

二、RDD编程 

1.RDD创建

1.1从集合中创建

1.2 从文件中读取数据集创建

1.3 从其RDD创建 

2.分区规则

2.1从集合创建RDD

2.2 从文件创建RDD

3.Transformation转换算子

3.1Value类型

1.map()映射

2.mapPartitions()以分区为单位执行Map

map()和mapPartitions()区别:

 3.mapPartitionsWithIndex()带分区号

 4.flatMap()扁平化，类似与炸裂函数

5.groupBy()分组   会走shuffle

6. filter()过滤   类似于where

 7.distinct()去重   走shuffle        

8.coalesce()合并分区

 9.repartition()重新分区，默认走shuffle

coalesce和repartition区别

3.2 Key-Value类型

要想使用Key-Value类型的算子首先需要使用特定的方法转换为PairRDD

1. mapValues()只对Value进行操作

 2.partitonBy() 按照key值重新分区

自定义分区器

3.groupByKey()按照K重新分组

4.reduceByKey()   reduceByKey()按照K聚合V，效果等于groupBy  +  mapValue

reduceByKey()和groupByKey()的区别

5.sortByKey()按照K进行排序

6.join()等同于sql里的内连接,关联上的要,关联不上的舍弃

 7.cogroup()类似于sql的全连接，但是在同一个RDD中对key聚合

 4.action行动算子

1.collect() 行动算子 ，以数组的形式返回数据集

2.count()行动算子，返回RDD中元素个数

  first()返回RDD中的第一个元素

take()返回由RDD前n个元素组成的数组

takeOrdered()返回该RDD排序后前n个元素组成的数组

countByKey()统计每种key的个数

 saveAsTextFile(path)保存成Text文件

saveAsObjectFile(path) 序列化成对象保存到文件

 foreach()遍历RDD中每一个元素

5.WordCount案例实操

1，本地调试

2.集群运行

 3.进阶案例

4.RDD序列化

4.1 

一、RDD的概念

1. RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象。

代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

2.RDD的五大特性

（1）RDD是由一系列partition组成(block块对应partition),textFile底层调用的是MR读取hdfs上的数据的方法默认一个block块对应一个split,split的大小和block大小一致,可以自己调整

（2）函数作用在每一个partition(split)上

（3）RDD之间有一系列的依赖关系(容错机制)

（4）分区器作用在K,V格式的RDD上

（5）RDD 提供一系列最佳的计算位置

二、RDD编程 

1.RDD创建

1.1从集合中创建

  1. package com.atguigu.createrdd;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.Arrays;
  8. import java.util.List;
  9. import java.util.function.Consumer;
  10.  
  11. public class Test01_List {
  12.     public static void main(String[] args) {
  13.         //创建配置对象
  14.         SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
  15.         //创建sparkContext
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //使用sc编写代码
  18.         JavaRDD<String> javaRDD = sc.parallelize(Arrays.asList("hello", "spark"));
  19.  
  20.         List<String> result = javaRDD.collect();
  21.         //遍历打印
  22.         for (String s : result) {
  23.             System.out.println(s);
  24.         }
  25.         //函数式写法
  26.         result.forEach(new Consumer<String>() {
  27.             @Override
  28.             public void accept(String s) {
  29.                 System.out.println("值为 " + s);
  30.             }
  31.         });
  32.  
  33.         
  34.         //lambda表达式：
  35.         //1.箭头左边给传入的参数起名字，不用写类型
  36.         //2.箭头右边编写方法体，如果方法题有多行，使用{}括起来，
  37.         //3. 如果有返回值  不写return  使用最后一行代码的结果作为返回值
  38.         result.forEach(s ->{
  39.                     System.out.println("hi");
  40.                     System.out.println("值为 " + s);
  41.                 });
  42.         
  43.         // lambda表达式进一步简化
  44.         result.forEach(System.out::println);
  45.         //关闭sc
  46.         sc.close();
  47.     }
  48. }

1.2 从文件中读取数据集创建

  1. package com.atguigu.createrdd;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.List;
  8. import java.util.function.Consumer;
  9.  
  10. public class Test02_FromFile {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaRDD<String> lineRDD = sc.textFile("input/2.txt");
  18.  
  19.         List<String> collect = lineRDD.collect();
  20.         //将结果保存到文件中   有一个文件就是对应一个分区
  21.         lineRDD.saveAsTextFile("output");
  22.  
  23.         collect.forEach(System.out::println);
  24.  
  25.         lineRDD.collect().forEach(new Consumer<String>() {
  26.             @Override
  27.             public void accept(String s) {
  28.                 System.out.println("---------");
  29.                 System.out.println(s);
  30.             }
  31.         });
  32.  
  33.         lineRDD. collect().forEach(System.out::println);
  34.         //关闭sc
  35.         sc.stop();
  36.     }
  37. }

1.3 从其RDD创建 

2.分区规则

关于分区都不会shuffle，一个shuffle会有一个读和写的过程，只有一个写不走shuffle

2.1从集合创建RDD

默认环境的核数，可自行设置参数

  1. package com.atguigu.createrdd;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.util.Utils;
  7.  
  8. import java.io.File;
  9. import java.util.Arrays;
  10.  
  11. public class Test03_ListPartition {
  12.     public static void main(String[] args) throws InterruptedException {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.         //默认的分区个数  环境核数
  19.         //可以手动填写参数修改分区
  20.         JavaRDD<Integer> integerJavaRDD = sc.parallelize(Arrays.asList(1,2,3,4,5),2);
  21.  
  22.         // 多余的数据会放到后面一个分区，比如现在5个元素，12在一个分区，345在另一个分区
  23.         // 将结果保存到文件中   有一个文件就是对应一个分区
  24.         Utils.deleteRecursively(new File("output"));
  25.         integerJavaRDD.saveAsTextFile("output");
  26.  
  27.         //关闭sc
  28.         sc.stop();
  29.     }
  30. }

2.2 从文件创建RDD

  1. package com.atguigu.createrdd;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. public class Test04_FilePartition {
  8.     public static void main(String[] args) {
  9.         //创建spark配置对象
  10.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  11.         //创建sc环境
  12.         JavaSparkContext sc = new JavaSparkContext(conf);
  13.         //编写代码
  14.         //默认填写最小分区数为2  min（2，环境的核数）
  15.         JavaRDD<String> stringJavaRDD = sc.textFile("input/1.txt",2);
  16.  
  17.         //spark的读文件使用了老版本的hadoop代码
  18.         // 具体的分区个数需要经过公式计算
  19.         // 首先获取文件的总长度  totalSize
  20.         // 计算平均长度  goalSize = totalSize / numSplits
  21.         // 获取块大小 128M
  22.         // 计算切分大小  splitSize = Math.max(minSize, Math.min(goalSize, blockSize));
  23.         // 最后使用splitSize  按照1.1倍原则切分整个文件   得到几个分区就是几个分区
  24.  
  25.         // 实际开发中   只需要看文件总大小 / 填写的分区数  和块大小比较  谁小拿谁进行切分
  26.         lineRDD.saveAsTextFile("output");
  27.  
  28.         // 数据会分配到哪个分区
  29.         // 如果切分的位置位于一行的中间  会在当前分区读完一整行数据
  30.  
  31.         stringJavaRDD.saveAsTextFile("output");
  32.  
  33.         //关闭sc
  34.         sc.stop();
  35.     }
  36. }

3.Transformation转换算子

3.1Value类型

1.map()映射

参数f是一个函数可以写作匿名子类，它可以接收一个参数。当某个RDD执行map方法时，会遍历该RDD中的每一个数据项，并依次应用f函数，从而产生一个新的RDD。即，这个新RDD中的每一个元素都是原来RDD中每一个元素依次应用f函数而得到的。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7. import scala.Tuple2;
  8.  
  9. import java.util.Arrays;
  10.  
  11. public class Test01_Map {
  12.     public static void main(String[] args) {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5), 2);
  19.  
  20.         //使用map转换
  21.         //匿名函数的写法
  22.         //填写的泛型  填一个为传入参数的类型   后一个为返回值的类型（同时作为下游RDD 的元素类型）
  23.         JavaRDD<Integer> mapRDD = intRDD.map(new Function<Integer, Integer>() {
  24.             @Override
  25.             public Integer call(Integer v1) throws Exception {
  26.                 return v1 * 2;
  27.             }
  28.         });
  29.         mapRDD. collect().forEach(System.out::println);
  30.         
  31.         //map是可以改变元素的类型的
  32.         JavaRDD<Tuple2<String, Integer>> tuple2JavaRDD1 = intRDD.map(new Function<Integer, Tuple2<String, Integer>>() {
  33.             @Override
  34.             public Tuple2<String, Integer> call(Integer v1) throws Exception {
  35.                     return new Tuple2<>("值为：", v1);
  36.             }
  37.         });
  38.  
  39.         JavaRDD<Tuple2<String, Integer>> tuple2JavaRDD = intRDD.map(new Function<Integer, Tuple2<String, Integer>>() {
  40.             @Override
  41.             public Tuple2<String, Integer> call(Integer v1) throws Exception {
  42.                 return new Tuple2<>("值为：", v1);
  43.             }
  44.         });
  45.  
  46.         intRDD. collect().forEach(System.out::println);
  47.  
  48.         //关闭sc
  49.         sc.stop();
  50.     }
  51. }

2.mapPartitions()以分区为单位执行Map

功能和map一样都是处理每个元素，Map是一次处理一个元素，而mapPartitions一次处理一个分区数据。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.FlatMapFunction;
  7. import org.apache.spark.api.java.function.Function;
  8. import org.apache.spark.util.Utils;
  9.  
  10. import java.io.File;
  11. import java.util.ArrayList;
  12. import java.util.Arrays;
  13. import java.util.Iterator;
  14.  
  15. public class Test02_MapPartitions {
  16.     public static void main(String[] args) throws InterruptedException {
  17.         //创建spark配置对象
  18.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  19.         //创建sc环境
  20.         JavaSparkContext sc = new JavaSparkContext(conf);
  21.         //编写代码
  22.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5,6,7,8,9), 4);
  23.         //int值 * 2
  24.         //第二个泛型返回结果RDD的元素类型
  25.         JavaRDD<Double> mapPartitionsRDD = intRDD.mapPartitions(new FlatMapFunction<Iterator<Integer>, Double>() {
  26.             @Override
  27.             public Iterator<Double> call(Iterator<Integer> integerIterator) throws Exception {
  28.                 //将一个分区的数据全部收集到一个迭代器中  一次性处理所有的数据
  29.                 System.out.println("mapPartitions方法调用");
  30.                 ArrayList<Double> doubles = new ArrayList<>();
  31.                 while (integerIterator.hasNext()) {
  32.                     Integer next = integerIterator.next();
  33.                     doubles.add(next.doubleValue() * 2);
  34.                 }
  35.                 return doubles.iterator();
  36.             }
  37.         });
  38.         mapPartitionsRDD. collect().forEach(System.out::println);
  39.         //map操作
  40.         JavaRDD<Double> mapRDD = intRDD.map(new Function<Integer, Double>() {
  41.             @Override
  42.             public Double call(Integer v1) throws Exception {
  43.                 System.out.println("map方法调用");
  44.                 return v1.doubleValue() * 2;
  45.             }
  46.         });
  47.         System.out.println("----------------------------");
  48.         mapRDD. collect().forEach(System.out::println);
  49.         //需求是:将奇数的数字删除，偶数*2
  50.         JavaRDD<Double> mapRDD1 = intRDD.map(new Function<Integer, Double>() {
  51.             @Override
  52.             public Double call(Integer v1) throws Exception {
  53.                 if (v1 % 2 == 1) {
  54.                     return null;
  55.                 } else {
  56.                     return v1.doubleValue() * 2;
  57.                 }
  58.             }
  59.         });
  60.         mapRDD1. collect().forEach(System.out::println);
  61.  
  62.         JavaRDD<Double> mapPartitionsRDD1 = intRDD.mapPartitions(new FlatMapFunction<Iterator<Integer>, Double>() {
  63.             @Override
  64.             public Iterator<Double> call(Iterator<Integer> integerIterator) throws Exception {
  65.                 ArrayList<Double> doubles = new ArrayList<>();
  66.                 while (integerIterator.hasNext()) {
  67.                     Integer next = integerIterator.next();
  68.                     if (next % 2 == 0) {
  69.                         doubles.add(next.doubleValue() * 2);
  70.                     }
  71.                 }
  72.                 return doubles.iterator();
  73.             }
  74.         });
  75.         System.out.println("--------------------------");
  76.         mapPartitionsRDD1. collect().forEach(System.out::println);
  77.         JavaRDD<Integer> intMapPartitionsRDD = intRDD.mapPartitions(new FlatMapFunction<Iterator<Integer>, Integer>() {
  78.             @Override
  79.             public Iterator<Integer> call(Iterator<Integer> integerIterator) throws Exception {
  80.                 ArrayList<Integer> integers = new ArrayList<>();
  81.                 while (integerIterator.hasNext()) {
  82.                     Integer next = integerIterator.next();
  83.                     if (next % 2 == 0) {
  84.                         integers.add(next * 2);
  85.                     }
  86.                 }
  87.                 return integers.iterator();
  88.             }
  89.         });
  90.  
  91.         Utils.deleteRecursively(new File("output"));//如果有output文件夹就会删除
  92.         intMapPartitionsRDD.saveAsTextFile("output");
  93.         //关闭sc
  94.         sc.stop();
  95.     }
  96. }

map()和mapPartitions()区别:

 map一次处理一个元素，要求元素个数一点是1对1，不能删除元素
mapPartitions一次处理一个分区的数据，可以删除元素
如果资源充足，可以使用mapPartitions对map进行优化

如下为两个的结果

 3.mapPartitionsWithIndex()带分区号

类似于mapPartitions，比mapPartitions多一个整数参数表示分区号

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function2;
  7. import scala.Tuple2;
  8.  
  9. import java.util.ArrayList;
  10. import java.util.Arrays;
  11. import java.util.Iterator;
  12.  
  13. public class Test03_MapPartitionsWithIndex {
  14.     public static void main(String[] args) {
  15.         //创建spark配置对象
  16.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  17.         //创建sc环境
  18.         JavaSparkContext sc = new JavaSparkContext(conf);
  19.         //编写代码
  20.         JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5), 2);
  21.         //函数逻辑和mapPartitions完全一致，只是传一个参数，多了一个分区号
  22.         //第一个参数表示分区号，最后要的类型是 
  23.         // 需求：int -> （分区号，int）
  24.         JavaRDD<Tuple2<Integer, Integer>> tuple2JavaRDD = parallelize.mapPartitionsWithIndex(new Function2<Integer, Iterator<Integer>, Iterator<Tuple2<Integer, Integer>>>() {
  25.             @Override
  26.             public Iterator<Tuple2<Integer, Integer>> call(Integer v1, Iterator<Integer> v2) throws Exception {
  27.                 //v1表示分区号，v2表示元素值
  28.                 ArrayList<Tuple2<Integer, Integer>> tuple2s = new ArrayList<>();
  29.                 while (v2.hasNext()) {
  30.                     Integer next = v2.next();
  31.                     tuple2s.add(new Tuple2<>(v1, next));
  32.                 }
  33.                 return tuple2s.iterator();
  34.             }
  35.         }, false);
  36.  
  37.         tuple2JavaRDD. collect().forEach(System.out::println);
  38.         //关闭sc
  39.         sc.stop();
  40.     }
  41. }

 4.flatMap()扁平化，类似与炸裂函数

与map操作类似，将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。

区别：在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.FlatMapFunction;
  7.  
  8. import java.util.ArrayList;
  9. import java.util.Arrays;
  10. import java.util.Iterator;
  11. import java.util.List;
  12.  
  13. public class Test04_FlatMap {
  14.     public static void main(String[] args) {
  15.         //创建spark配置对象
  16.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  17.         //创建sc环境
  18.         JavaSparkContext sc = new JavaSparkContext(conf);
  19.         //编写代码
  20.         //集合嵌套的格式
  21.         ArrayList<List<Integer>> lists = new ArrayList<>();
  22.         lists.add(Arrays.asList(1, 2, 3));
  23.         lists.add(Arrays.asList(4, 5, 6));
  24.         lists.add(Arrays.asList(7, 8, 9));
  25.         JavaRDD<List<Integer>> listRDD = sc.parallelize(lists,2);
  26.         //flatMap()扁平化类似于hql的炸裂函数，就是将
  27.         JavaRDD<Integer> flatRDD = listRDD.flatMap(new FlatMapFunction<List<Integer>, Integer>() {
  28.             @Override
  29.             public Iterator<Integer> call(List<Integer> integers) throws Exception {
  30.                 return integers.iterator();
  31.             }
  32.         });
  33.         flatRDD. collect().forEach(System.out::println);
  34.  
  35.         //实际场景
  36.         //文件里面是hello world hi nike james lakers list
  37.         JavaRDD<String> lineRDD = sc.textFile("input/1.txt");
  38.  
  39.         JavaRDD<String> flatRDD1 = lineRDD.flatMap(new FlatMapFunction<String, String>() {
  40.  
  41.             @Override
  42.             public Iterator<String> call(String s) throws Exception {
  43.                 String[] s1 = s.split(" ");//通过字符串切割出来数组
  44.                 List<String> strings = Arrays.asList(s1);//数组转化为集合
  45.                 return strings.iterator();
  46.             }
  47.         });
  48.         flatRDD1. collect().forEach(System.out::println);
  49.         //关闭sc
  50.         sc.stop();
  51.     }
  52. }

5.groupBy()分组   会走shuffle

分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.Function;
  8. import org.apache.spark.util.Utils;
  9.  
  10. import java.io.File;
  11. import java.util.Arrays;
  12.  
  13. public class Test05_GroupBy {
  14.     public static void main(String[] args) throws InterruptedException {
  15.         //创建spark配置对象
  16.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  17.         //创建sc环境
  18.         JavaSparkContext sc = new JavaSparkContext(conf);
  19.         //编写代码
  20.         //需求：奇偶进行分组
  21.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(7, 8, 10, 11, 25, 16), 2);
  22.         //泛型1为传入参数，泛型2为分组依据
  23.         //最终返回类型：
  24.         // (根据分区依据计算的值（可以是任意类型的值），[根据分组依据将符合同一种的元素范进一个集合中])
  25.         JavaPairRDD<Integer, Iterable<Integer>> groupByRDD = intRDD.groupBy(new Function<Integer, Integer>() {
  26.             @Override
  27.             public Integer call(Integer integer) throws Exception {
  28.                 return integer % 3;
  29.             }
  30.         });
  31.         groupByRDD. collect().forEach(System.out::println);
  32.         //标记类型是可以改变的
  33.         JavaPairRDD<String, Iterable<Integer>> groupByRDD1 = intRDD.groupBy(new Function<Integer, String>() {
  34.             @Override
  35.             public String call(Integer integer) throws Exception {
  36.                 return integer % 2 == 0 ? "偶数" : "奇数";
  37.             }
  38.         });
  39.         JavaPairRDD<Integer, Iterable<Integer>> groupByRDD2 = intRDD.groupBy(new Function<Integer, Integer>() {
  40.             @Override
  41.             public Integer call(Integer integer) throws Exception {
  42.                 return integer % 2 == 0 ? 2 : 5;
  43.             }
  44.         });
  45.         groupByRDD1. collect().forEach(System.out::println);
  46.         intRDD.groupBy((Function<Integer, String>) integer -> integer % 2 == 0 ? "偶数" : "奇数"). collect().forEach(System.out::println);
  47.         Utils.deleteRecursively(new File("output"));
  48.         groupByRDD.saveAsTextFile("output");
  49.         //关闭sc
  50.         sc.stop();
  51.     }
  52. }

6. filter()过滤   类似于where

接收一个返回值为布尔类型的函数作为参数。当某个RDD调用filter方法时，会对该RDD中每一个元素应用f函数，如果返回值类型为true，则该元素会被添加到新的RDD中

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7.  
  8. import java.util.Arrays;
  9.  
  10. public class Test06_Filter {
  11.     public static void main(String[] args) throws InterruptedException {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 2);
  18.         //一个元素过滤一次
  19.         //相当于hql中的where，返回值只能是Boolean类型
  20.         JavaRDD<Integer> filterRDD = intRDD.filter(new Function<Integer, Boolean>() {
  21.             @Override
  22.             public Boolean call(Integer integer) throws Exception {
  23.                 System.out.println("调用方法了");
  24.                 return integer % 2 == 0;
  25.             }
  26.         });
  27.         filterRDD. collect().forEach(System.out::println);
  28.         //关闭sc
  29.         sc.stop();
  30.     }
  31. }
  32.  
  33. //代码结果
  34. //调用方法了
  35. //调用方法了
  36. //调用方法了
  37. //调用方法了
  38. //调用方法了
  39. //2
  40. //4
  41. //6

 7.distinct()去重   走shuffle        

对内部的元素去重，并将去重后的元素放到新的RDD中。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.Arrays;
  8.  
  9. public class Test07_Distinct {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         //创建spark配置对象
  12.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  13.         //创建sc环境
  14.         JavaSparkContext sc = new JavaSparkContext(conf);
  15.         //编写代码
  16.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 3, 2, 4, 5, 6, 7, 5, 3), 2);
  17.         //使用单个的distinct去重走shuffle的，去重的方式就是hash set
  18.         // 所有数据要进内存，效率很高但是肯能内存溢出，oom错误
  19.         //按照group by 进行分布式去重，效率比较低，需要走shuffle，不会内存溢出
  20.         JavaRDD<Integer> distinct = intRDD.distinct();
  21.         distinct. collect().forEach(System.out::println);
  22.  
  23.         //关闭sc
  24.         sc.stop();
  25.     }
  26. }

DAG有向无环图，中走shuffle的话正常是shuffle后有方法，但是这个distinct在shuffle前后都有是因为底层调了别的方法，map(),reduceByKey, 等，前面那个是distinct其实运行的是map()

 

8.coalesce()合并分区

Coalesce算子包括：配置执行Shuffle和配置不执行Shuffle两种方式。

缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.Arrays;
  8.  
  9. public class Test08_Coalesce {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         //创建spark配置对象
  12.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  13.         //创建sc环境
  14.         JavaSparkContext sc = new JavaSparkContext(conf);
  15.         //编写代码
  16.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7), 7);
  17.         //默认不走shuffle，可以设置参数Boolean型参数，false为不走shuffle
  18.         JavaRDD<Integer> coalesceRDD = intRDD.coalesce(2);
  19.         JavaRDD<Integer> coalesceRDD1 = intRDD.coalesce(2, false);
  20. //        Utils.deleteRecursively(new File("output"));
  21. //        coalesceRDD.saveAsTextFile("output");
  22.         coalesceRDD. collect().forEach(System.out::println);
  23.         //关闭sc
  24.         sc.stop();
  25.     }
  26. }

 9.repartition()重新分区，默认走shuffle

该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的RDD转换为分区数多的RDD，repartition操作都可以完成，因为无论如何都会经shuffle过程。分区规则不是hash，因为平时使用的分区都是按照hash来实现的，repartition一般是对hash的结果不满意，想要打散重新分区。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.Arrays;
  8.  
  9. public class Test09_Repartition {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         //创建spark配置对象
  12.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  13.         //创建sc环境
  14.         JavaSparkContext sc = new JavaSparkContext(conf);
  15.         //编写代码
  16.         JavaRDD<Integer> intRDD  = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7), 2);
  17.         // 0 -> 1,2  1 -> 3,4  2 -> 5,6,7
  18.         // 分区的规则就是打散  尽量把之前同一个分区的数据  放到不同的分区里面
  19.         // 如果按照默认的hash分区  会发生数据倾斜   repartition重新打散分区
  20.         // 0 ->     1 -> 1      2-> 2,5      3 -> 3,6     4 -> 4,7
  21.         JavaRDD<Integer> repartitionRDD = intRDD.repartition(3);
  22.         repartitionRDD. collect().forEach(System.out::println);
  23.         //关闭sc
  24.         sc.stop();
  25.     }
  26. }

coalesce和repartition区别

1）coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。

2）repartition实际上是调用的coalesce，进行shuffle。源码如下：

  1. def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
  2.     coalesce(numPartitions, shuffle = true)
  3. }

3）coalesce一般为缩减分区，如果扩大分区，不使用shuffle是没有意义的，repartition扩大分区执行shuffle。

 10. sortBy()排序    走shuffle

该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为正序排列。排序后新产生的RDD的分区数与原RDD的分区数一致。Spark的排序结果是全局有序。

  1. package com.atguigu.value;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7.  
  8. import java.util.Arrays;
  9.  
  10. public class Test10_SortBy {
  11.     public static void main(String[] args) throws InterruptedException {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(3, 3, 4, 5, 1, 7, 6, 8, 4, 2), 2);
  18.         //(1)泛型为以谁作为标准排序  (2) true为正序  (3) 排序之后的分区个数
  19.         //spark是全局排序
  20.         //使用特殊分区器，range分区器
  21.         //做法：先抽样，抽样中选出最大最小值，
  22.         // 当分区为2的时候，取最大值最小值的中位数，为3个的时候三分数
  23.         //比中位数大的分一个分区，小的放另一个，然后分桶，分桶后排序
  24.         JavaRDD<Integer> sortByRDD = intRDD.sortBy(new Function<Integer, Integer>() {
  25.             @Override
  26.             public Integer call(Integer v1) throws Exception {
  27.                 return v1;
  28.             }
  29.         }, true, 3);
  30.         sortByRDD.collect().forEach(System.out::println);
  31. //        Utils.deleteRecursively(new File("output"));
  32. //        sortByRDD.saveAsTextFile("output");
  33.         Thread.sleep(120000);
  34.         //关闭sc
  35.         sc.stop();
  36.     }
  37. }

3.2 Key-Value类型

要想使用Key-Value类型的算子首先需要使用特定的方法转换为PairRDD

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.PairFunction;
  8. import scala.Tuple2;
  9.  
  10. import java.util.Arrays;
  11.  
  12. public class Test01_PairRDD {
  13.     public static void main(String[] args) {
  14.         //创建spark配置对象
  15.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  16.         //创建sc环境
  17.         JavaSparkContext sc = new JavaSparkContext(conf);
  18.         //编写代码
  19.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
  20.  
  21.         //使用mapToPair()方法将JavaRDD转化为javaPairRDD
  22.         //第一个泛型表示当前RDD的元素类型
  23.         //第二个泛型表示k的类型
  24.         //第三个表示value的类型
  25.         JavaPairRDD<Integer, Integer> pairRDD = intRDD.mapToPair(new PairFunction<Integer, Integer, Integer>() {
  26.             @Override
  27.             public Tuple2<Integer, Integer> call(Integer integer) throws Exception {
  28.                 return new Tuple2<>(integer, integer);
  29.             }
  30.         });
  31.         pairRDD. collect().forEach(System.out::println);
  32.         //直接创建JavaPairRDD的对象
  33.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("a", 1), new Tuple2<>("b", 2)));
  34.         pairRDD1. collect().forEach(System.out::println);
  35.         //关闭sc
  36.         sc.stop();
  37.     }
  38. }

1. mapValues()只对Value进行操作

针对于(K,V)形式的类型只对V进行操作 

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7. import scala.Tuple2;
  8.  
  9. import java.util.Arrays;
  10.  
  11. public class Test02_MapValues {
  12.     public static void main(String[] args) {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.         //javaRDD不能直接调mapValues
  19.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("a", 1), new Tuple2<>("b", 2)));
  20.         JavaPairRDD<String, String> mapValuesRDD = pairRDD1.mapValues(new Function<Integer, String>() {
  21.             @Override
  22.             public String call(Integer integer) throws Exception {
  23.                 return integer + "|||";
  24.             }
  25.         });
  26.         mapValuesRDD. collect().forEach(System.out::println);
  27.         //关闭sc
  28.         sc.stop();
  29.     }
  30. }

 2.partitonBy() 按照key值重新分区

将RDD<K,V>中的K按照指定Partitioner重新进行分区；

如果原有的RDD和新的RDD分区器是一致的话就不进行分区，否则会产生Shuffle过程。

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.HashPartitioner;
  4. import org.apache.spark.SparkConf;
  5. import org.apache.spark.api.java.JavaPairRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.util.Utils;
  8. import scala.Tuple2;
  9.  
  10. import java.io.File;
  11. import java.util.Arrays;
  12.  
  13. public class Test03_PartitionBy {
  14.     public static void main(String[] args) throws InterruptedException {
  15.         //创建spark配置对象
  16.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  17.         //创建sc环境
  18.         JavaSparkContext sc = new JavaSparkContext(conf);
  19.         //编写代码
  20.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("a", 1), new Tuple2<>("b", 2)));
  21.         //填写分区器进行分区，实现partitioner的类只有三个：
  22.         // Hash、Range(一般不用，都是系统自己调用)、Python
  23.         //一般都用HashPartitioner
  24.         JavaPairRDD<String, Integer> partitionByRDD = pairRDD1.partitionBy(new HashPartitioner(2));
  25.         JavaPairRDD<Integer, Integer> pairRDD2 = sc.parallelizePairs(Arrays.asList(new Tuple2<>(2, 1), new Tuple2<>(2, 2),new Tuple2<>(4, 3),new Tuple2<>(6, 4)));
  26.         // 如果key为int值  hash值就是他自身  % 分区的个数
  27.         JavaPairRDD<Integer, Integer> partitionByRDD1 = pairRDD2.partitionBy(new HashPartitioner(2));
  28.         Utils.deleteRecursively(new File("output"));
  29.         partitionByRDD1.saveAsTextFile("output");
  30.         //关闭sc
  31.         sc.stop();
  32.     }
  33. }

自定义分区器

要实现自定义分区器，需要继承org.apache.spark.Partitioner类，并实现下面三个方法。

（1）numPartitions: Int:返回创建出来的分区数。

（2）getPartition(key: Any): Int:返回给定键的分区编号（0到numPartitions-1）。

（3）equals():Java 判断相等性的标准方法。这个方法的实现非常重要，Spark需要用这个方法来检查你的分区器对象是否和其他分区器实例相同，这样Spark才可以判断两个RDD的分区方式是否相同。

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.Partitioner;
  4.  
  5. public class CustomPartitioner extends Partitioner {
  6.     Integer num = 0;
  7.     public CustomPartitioner( int n ) {
  8.         num = n;
  9.     }
  10.  
  11.     @Override
  12.     public int numPartitions() {
  13.         return num;
  14.     }
  15.  
  16.     @Override
  17.     public int getPartition(Object key) {
  18.         //根据需求，但是必须按照key
  19.         // 如果是int值  按照对num取模
  20.         // 不为int值 发送到0号分区
  21.         if (key instanceof Integer) {
  22.             return (Integer) key % num;
  23.         }
  24.         return 0;
  25.     }
  26. }

使用该分区器

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.util.Utils;
  7. import scala.Tuple2;
  8.  
  9. import java.io.File;
  10. import java.util.Arrays;
  11.  
  12. public class Test04_CustomPartitioner {
  13.     public static void main(String[] args) {
  14.         //创建spark配置对象
  15.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  16.         //创建sc环境
  17.         JavaSparkContext sc = new JavaSparkContext(conf);
  18.         //编写代码
  19.         JavaPairRDD<Integer, Integer> pairRDD2 = sc.parallelizePairs(Arrays.asList(new Tuple2<>(1, 1), new Tuple2<>(2, 2),new Tuple2<>(4, 3),new Tuple2<>(6, 4)));
  20.  
  21.         JavaPairRDD<Integer, Integer> partitionByRDD1 = pairRDD2.partitionBy(new CustomPartitioner(3));
  22.         Utils.deleteRecursively(new File("output"));
  23.         partitionByRDD1.saveAsTextFile("output");
  24.         //关闭sc
  25.         sc.stop();
  26.     }
  27. }

3.groupByKey()按照K重新分组

groupByKey对每个key进行操作，但只生成一个seq，并不进行聚合。

该操作可以指定分区器或者分区数（默认使用HashPartitioner）

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7. import scala.Tuple2;
  8.  
  9. import java.util.Arrays;
  10.  
  11. public class Test05_GroupByKey {
  12.     public static void main(String[] args) throws InterruptedException {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("hello", 1), new Tuple2<>("world", 2),new Tuple2<>("hello", 11), new Tuple2<>("world", 21),new Tuple2<>("hello", 14), new Tuple2<>("world", 22)));
  19.         JavaPairRDD<String, Iterable<Integer>> groupByKeyRDD = pairRDD1.groupByKey();
  20.         groupByKeyRDD. collect().forEach(System.out::println);
  21.  
  22.         JavaPairRDD<String, Integer> mapValuesRDD = groupByKeyRDD.mapValues(new Function<Iterable<Integer>, Integer>() {
  23.             @Override
  24.             public Integer call(Iterable<Integer> v1) throws Exception {
  25.                 Integer sum = 0;
  26.                 for (Integer integer : v1) {
  27.                     sum += integer;
  28.                 }
  29.                 return sum;
  30.             }
  31.         });
  32.         mapValuesRDD. collect().forEach(System.out::println);
  33.         //关闭sc
  34.         sc.stop();
  35.     }
  36. }

4.reduceByKey()   reduceByKey()按照K聚合V，效果等于groupBy  +  mapValue

该操作可以将RDD[K,V]中的元素按照相同的K对V进行聚合。其存在多种重载形式，还可以设置新RDD的分区数。

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function2;
  7. import scala.Tuple2;
  8.  
  9. import java.util.Arrays;
  10.  
  11. public class Test06_ReduceByKey {
  12.     public static void main(String[] args) {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.  
  19.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("hello", 1), new Tuple2<>("world", 2),new Tuple2<>("hello", 11), new Tuple2<>("world", 21),new Tuple2<>("hello", 14), new Tuple2<>("world", 22)));
  20.         //reduceByKey  = groupByKey  +   mapValues
  21.         JavaPairRDD<String, Integer> reduceByKeyRDD = pairRDD1.reduceByKey(new Function2<Integer, Integer, Integer>() {
  22.             @Override
  23.             //v1为当前累计的结果，v2为当前的元素值
  24.             //第一个sum的值取的就是第一个elem，elem是从第二个开始循环的
  25.             public Integer call(Integer sum, Integer elem) throws Exception {
  26.                 // 第一次计算的时候  sum = 第一个元素  elem = 第二个元素
  27.                 return sum += elem;
  28.             }
  29.         });
  30.         reduceByKeyRDD. collect().forEach(System.out::println);
  31.  
  32.         //关闭sc
  33.         sc.stop();
  34.     }
  35. }

reduceByKey()和groupByKey()的区别

1）reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[K,V]。

2）groupByKey：按照key进行分组，直接进行shuffle。

3）开发指导：在不影响业务逻辑的前提下，优先选用reduceByKey。求和操作不影响业务逻辑，求平均值影响业务逻辑。影响业务逻辑时建议先对数据类型进行转换再合并。

reduceByKey()求平均值

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7. import org.apache.spark.api.java.function.Function2;
  8. import scala.Tuple2;
  9.  
  10. import java.util.Arrays;
  11.  
  12. public class Test07_reduceByKeyAvg {
  13.     public static void main(String[] args) {
  14.         //创建spark配置对象
  15.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  16.         //创建sc环境
  17.         JavaSparkContext sc = new JavaSparkContext(conf);
  18.         //编写代码
  19.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("hello", 1), new Tuple2<>("world", 2),new Tuple2<>("hello", 11), new Tuple2<>("world", 21),new Tuple2<>("hello", 14), new Tuple2<>("world", 22)),2);
  20.         //如果需要求平均值  需要两个数据  (sum,count)
  21.         //value -> (value,1)  表示当前的数据和为value  值的个数为1
  22.         JavaPairRDD<String, Tuple2<Integer, Integer>> tuple2RDD = pairRDD1.mapValues(new Function<Integer, Tuple2<Integer, Integer>>() {
  23.             @Override
  24.             public Tuple2<Integer, Integer> call(Integer v1) throws Exception {
  25.                 return new Tuple2<>(v1, 1);
  26.             }
  27.         });
  28.         tuple2RDD. collect().forEach(System.out::println);
  29.         System.out.println("---------------------");
  30.         // 之后使用reduceByKey聚合数据
  31.         JavaPairRDD<String, Tuple2<Integer, Integer>> pairRDD = tuple2RDD.reduceByKey(new Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
  32.             @Override
  33.             public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> res, Tuple2<Integer, Integer> elem) throws Exception {
  34.                 // 将sum值相加  同时将count值相加
  35.                 return new Tuple2<>(res._1 + elem._1, res._2 + elem._2);
  36.             }
  37.         });
  38.         pairRDD. collect().forEach(System.out::println);
  39.         System.out.println("---------------");
  40.  
  41.         // 对value值使用sum / count
  42.         JavaPairRDD<String, Double> resultRDD = pairRDD.mapValues(new Function<Tuple2<Integer, Integer>, Double>() {
  43.             @Override
  44.             public Double call(Tuple2<Integer, Integer> v1) throws Exception {
  45.                 return v1._1.doubleValue() / v1._2;
  46.             }
  47.         });
  48.         resultRDD. collect().forEach(System.out::println);
  49.  
  50.         //关闭sc
  51.         sc.stop();
  52.     }
  53. }

5.sortByKey()按照K进行排序

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.PairFunction;
  7. import scala.Tuple2;
  8.  
  9. import java.util.Arrays;
  10.  
  11. public class Test08_SortByKey {
  12.     public static void main(String[] args) {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.         JavaPairRDD<String, Integer> pairRDD = sc.parallelizePairs(Arrays.asList(new Tuple2<>("hello", 1), new Tuple2<>("world", 2),new Tuple2<>("hi", 11), new Tuple2<>("you", 21),new Tuple2<>("i", 14), new Tuple2<>("ok", 22)),2);
  19.         JavaPairRDD<String, Integer> sortByKeyRDD = pairRDD.sortByKey();
  20.         sortByKeyRDD. collect().forEach(System.out::println);
  21.  
  22.         //需求：按照后面数字大小排序
  23.         //把需要排序的标记放在key的位置上
  24.         JavaPairRDD<Integer, Tuple2<String, Integer>> pairRDD1 = pairRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, Tuple2<String, Integer>>() {
  25.             @Override
  26.             public Tuple2<Integer, Tuple2<String, Integer>> call(Tuple2<String, Integer> v1) throws Exception {
  27.                 return new Tuple2<>(v1._2, v1);
  28.             }
  29.         });
  30.         System.out.println("-----------");
  31.         pairRDD1. collect().forEach(System.out::println);
  32.  
  33.         JavaPairRDD<Integer, Tuple2<String, Integer>> pairRDD2 = pairRDD1.sortByKey();
  34.  
  35.         JavaPairRDD<String, Integer> resultRDD = pairRDD2.mapToPair(new PairFunction<Tuple2<Integer, Tuple2<String, Integer>>, String, Integer>() {
  36.             @Override
  37.             public Tuple2<String, Integer> call(Tuple2<Integer, Tuple2<String, Integer>> v1) throws Exception {
  38.                 return v1._2;
  39.             }
  40.         });
  41.         System.out.println("-------------");
  42.         resultRDD. collect().forEach(System.out::println);
  43.         //关闭sc
  44.         sc.stop();
  45.     }
  46. }

6.join()等同于sql里的内连接,关联上的要,关联不上的舍弃

在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import scala.Tuple2;
  7.  
  8. import java.util.Arrays;
  9.  
  10. public class Test09_Join {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.  
  18.         JavaPairRDD<Integer, String> pairRDD = sc.parallelizePairs(Arrays.asList(new Tuple2<>(1, "a"), new Tuple2<>(2, "b"), new Tuple2<>(4, "m")));
  19.  
  20.         JavaPairRDD<Integer, String> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>(1, "p"), new Tuple2<>(2, "o"), new Tuple2<>(4, "i"), new Tuple2<>(6, "i")));
  21.         // 相当于等值连接  rdd1.k = rdd2.k
  22.         JavaPairRDD<Integer, Tuple2<String, String>> joinRDD = pairRDD.join(pairRDD1);
  23.         joinRDD. collect().forEach(System.out::println);
  24.         //关闭sc
  25.         sc.stop();
  26.     }
  27. }

 7.cogroup()类似于sql的全连接，但是在同一个RDD中对key聚合

在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD。

操作两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。

  1. package com.atguigu.keyvalue;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import scala.Tuple2;
  7.  
  8. import java.util.Arrays;
  9.  
  10. public class Test10_Cogroup {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaPairRDD<Integer, String> pairRDD = sc.parallelizePairs(Arrays.asList(new Tuple2<>(1, "a"), new Tuple2<>(2, "b"), new Tuple2<>(4, "m"),new Tuple2<>(1, "f")));
  18.         JavaPairRDD<Integer, String> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>(1, "p"), new Tuple2<>(2, "o"), new Tuple2<>(4, "i"), new Tuple2<>(6, "i")));
  19.         JavaPairRDD<Integer, Tuple2<Iterable<String>, Iterable<String>>> result = pairRDD.cogroup(pairRDD1);
  20.         // rdd1.k = rdd2.k
  21.         // (k,(集合[rdd1的元素,],集合[rdd2的元素]))
  22.         result. collect().forEach(System.out::println);
  23.         //关闭sc
  24.         sc.stop();
  25.     }
  26. }

 4.action行动算子

1.collect() 行动算子 ，以数组的形式返回数据集

在驱动程序中，以数组Array的形式返回数据集的所有元素。所有的数据都会被拉取到Driver端，容易造成内存溢出，慎用。collect是按照分区的编号从0号开始一次拉取到driver

  1. package com.atguigu.action;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.Arrays;
  8. import java.util.List;
  9.  
  10. public class Test01_Collect {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 2);
  18.         // 按照分区的编号从0号分区开始拉取数据到driver
  19.         // 实际开发不会使用collect   因为会将所有的数据都拉取到driver端  容易内存溢出
  20.         List<Integer> collect = intRDD.collect();
  21.         collect.forEach(a -> System.out.println(a));
  22.  
  23.         //关闭sc
  24.         sc.stop();
  25.     }
  26. }

2.count()行动算子，返回RDD中元素个数

  first()返回RDD中的第一个元素

take()返回由RDD前n个元素组成的数组

takeOrdered()返回该RDD排序后前n个元素组成的数组

  1. package com.atguigu.action;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6.  
  7. import java.util.Arrays;
  8. import java.util.List;
  9.  
  10. public class Test02_Count {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaRDD<Integer> intRDD = sc.parallelize(Arrays.asList(4, 5, 1, 2, 3), 6);
  18.         long count = intRDD.count();
  19.         System.out.println(count);//结果为5
  20.     
  21.         // first 返回0号分区的第一个元素  如果没有顺位往下
  22.         Integer first = intRDD.first();
  23.         System.out.println(first);//结果为4
  24.  
  25.         // take取前n个元素
  26.         List<Integer> take = intRDD.take(2);
  27.         System.out.println(take);//[4,5]
  28.  
  29.         List<Integer> takeOrderedRDD = intRDD.takeOrdered(2);
  30.         System.out.println(takeOrderedRDD);//[1,2]正序排序后取前两个
  31.         //关闭sc
  32.         sc.stop();
  33.     }
  34. }

countByKey()统计每种key的个数

 saveAsTextFile(path)保存成Text文件

功能说明：将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本

saveAsObjectFile(path) 序列化成对象保存到文件

功能说明：用于将RDD中的元素序列化成对象，存储到文件中

  1. package com.atguigu.action;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import scala.Tuple2;
  7.  
  8. import java.util.Arrays;
  9. import java.util.Map;
  10.  
  11. public class Test03_CountByKey {
  12.     public static void main(String[] args) {
  13.         //创建spark配置对象
  14.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  15.         //创建sc环境
  16.         JavaSparkContext sc = new JavaSparkContext(conf);
  17.         //编写代码
  18.         JavaPairRDD<String, Integer> pairRDD1 = sc.parallelizePairs(Arrays.asList(new Tuple2<>("hello", 10), new Tuple2<>("world", 20),new Tuple2<>("hello", 10), new Tuple2<>("world", 20),new Tuple2<>("hello", 10), new Tuple2<>("world", 20)),2);
  19.         Map<String, Long> countByKeyRDD = pairRDD1.countByKey();
  20.         System.out.println(countByKeyRDD);//{hello=3, world=3}
  21.         // 保存为文本文件  能够直接读懂
  22.         pairRDD1.saveAsTextFile("output");
  23.         //保存为对象格式，无法读懂
  24.         pairRDD1.saveAsObjectFile("output1");
  25.         //关闭sc
  26.         sc.stop();
  27.     }
  28. }

 foreach()遍历RDD中每一个元素

  1. package com.atguigu.action;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.VoidFunction;
  7.  
  8. import java.util.Arrays;
  9.  
  10. public class Test04_Foreach {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 6);
  18.  
  19.         parallelize.foreach(new VoidFunction<Integer>() {
  20.             @Override
  21.             public void call(Integer integer) throws Exception {
  22.                 System.out.println(integer);
  23.             }
  24.         });//3,5,4,6,2,1
  25.  
  26.         //关闭sc
  27.         sc.stop();
  28.     }
  29. }

5.WordCount案例实操

1，本地调试

本地Spark程序调试需要使用Local提交模式，即将本机当做运行环境，Master和Worker都为本机。运行时直接加断点调试即可。

2.集群运行

  1. package com.atguigu;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.FlatMapFunction;
  8. import org.apache.spark.api.java.function.Function2;
  9. import org.apache.spark.api.java.function.PairFlatMapFunction;
  10. import org.apache.spark.api.java.function.PairFunction;
  11. import scala.Tuple2;
  12.  
  13. import java.util.ArrayList;
  14. import java.util.Arrays;
  15. import java.util.Iterator;
  16.  
  17. public class WordCount {
  18.     public static void main(String[] args) {
  19.         //创建spark配置对象
  20.         SparkConf conf = new SparkConf().setAppName("core").setMaster("yarn");
  21.         //创建sc环境
  22.         JavaSparkContext sc = new JavaSparkContext(conf);
  23.         //编写代码
  24.         JavaRDD<String> javaRDD = sc.textFile(args[0]);
  25.  
  26.         //将读进来的每一行数据进行切分，然后同过flatMap进行打散
  27.         JavaRDD<String> wordRDD = javaRDD.flatMap(new FlatMapFunction<String, String>() {
  28.             @Override
  29.             public Iterator<String> call(String s) throws Exception {
  30.                 String[] s1 = s.split(" ");
  31.                 return Arrays.asList(s1).iterator();
  32.             }
  33.         });
  34.  
  35.         //将javaRDD转化为JavaPairRDD
  36.         JavaPairRDD<String, Integer> wordPairRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {
  37.             @Override
  38.             public Tuple2<String, Integer> call(String s) throws Exception {
  39.                 return new Tuple2<>(s, 1);
  40.             }
  41.         });
  42.  
  43.         //等同于flatMap + mapToPair
  44.         JavaPairRDD<String, Integer> wordPairRDD1 = wordRDD.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
  45.             @Override
  46.             public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
  47.                 ArrayList<Tuple2<String, Integer>> tuple2s = new ArrayList<>();
  48.                 String[] words = s.split(" ");
  49.                 for (String word : words) {
  50.                     tuple2s.add(new Tuple2<>(word, 1));
  51.                 }
  52.                 return tuple2s.iterator();
  53.             }
  54.         });
  55.  
  56.         //使用reduceByKey()聚合
  57.         JavaPairRDD<String, Integer> result = wordPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
  58.             @Override
  59.             public Integer call(Integer v1, Integer v2) throws Exception {
  60.                 return v1 + v2;
  61.             }
  62.         });
  63.  
  64.         result.saveAsTextFile(args[1]);
  65.  
  66.         //关闭sc
  67.         sc.stop();
  68.     }
  69. }

进行打包

在HDFS上创建上传的路径，将需要WordCount的文件上传上去

然后执行（记得启动Hadoop）

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \

--class com.atguigu.spark.WordCount \

--master yarn \

./WordCount.jar \

/input \

/output

 3.进阶案例

（1）需求：省份广告被点击Top3

（2）数据准备：时间戳，省份，城市，用户，广告，中间字段使用空格分割。

  1. package com.atguigu;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.Function;
  8. import org.apache.spark.api.java.function.Function2;
  9. import org.apache.spark.api.java.function.PairFunction;
  10. import scala.Tuple2;
  11.  
  12. import java.util.ArrayList;
  13. import java.util.Map;
  14. import java.util.TreeMap;
  15.  
  16. public class Top3 {
  17.     public static void main(String[] args) {
  18.         //创建spark配置对象
  19.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  20.         //创建sc环境
  21.         JavaSparkContext sc = new JavaSparkContext(conf);
  22.         //编写代码
  23.         JavaRDD<String> lineRDD = sc.textFile("input/agent.log");
  24.  
  25.         // 将一整行字符串  1516609143867 山西 7 64 A
  26.         // 转换为   key为(省份,广告)  value 为 1
  27.         JavaPairRDD<Tuple2<String, String>, Integer> tuple2RDD = lineRDD.mapToPair(new PairFunction<String, Tuple2<String, String>, Integer>() {
  28.             @Override
  29.             public Tuple2<Tuple2<String, String>, Integer> call(String s) throws Exception {
  30.                 String[] s1 = s.split(" ");
  31.                 Tuple2<String, String> key = new Tuple2<>(s1[1], s1[4]);
  32.                 return new Tuple2<>(key, 1);
  33.             }
  34.         });
  35.  
  36.         //将省份广告id相同的聚合
  37.         JavaPairRDD<Tuple2<String, String>, Integer> reduceByKeyRDD = tuple2RDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
  38.             @Override
  39.             public Integer call(Integer v1, Integer v2) throws Exception {
  40.                 return v1 + v2;
  41.             }
  42.         });// 已经完成省份+广告的点击次数统计   ((省份,广告),sum)
  43.  
  44.         // 转换格式  将省份作为key   value为 (广告,sum)  然后聚合
  45.         JavaPairRDD<String, Tuple2<String, Integer>> stringTuple2JavaPairRDD = reduceByKeyRDD.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Integer>, String, Tuple2<String, Integer>>() {
  46.             @Override
  47.             public Tuple2<String, Tuple2<String, Integer>> call(Tuple2<Tuple2<String, String>, Integer> v1) throws Exception {
  48.                 String province = v1._1._1;
  49.                 String id = v1._1._2;
  50.                 Integer sum = v1._2;
  51.                 return new Tuple2<>(province, new Tuple2<>(id, sum));
  52.             }
  53.         });
  54.  
  55.         //将同一个省份的数据聚合
  56.         JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> provinceRDD = stringTuple2JavaPairRDD.groupByKey();
  57.  
  58.         //找出单个省份点击次数的top3
  59.         JavaPairRDD<String, ArrayList<Tuple2<String, Integer>>> result = provinceRDD.mapValues(new Function<Iterable<Tuple2<String, Integer>>, ArrayList<Tuple2<String, Integer>>>() {
  60.             @Override
  61.             public ArrayList<Tuple2<String, Integer>> call(Iterable<Tuple2<String, Integer>> v1) throws Exception {
  62.                 //将二元组的数据按照value进行排序
  63.                 //创建treeMap，把需要排序的当作key
  64.                 TreeMap<Integer, String> treeMap = new TreeMap<>();
  65.                 for (Tuple2<String, Integer> tuple2 : v1) {
  66.                     Integer key = tuple2._2;
  67.                     if (treeMap.containsKey(key)) {
  68.                         //当treeMap中存在相同次数的广告id
  69.                         //将广告id进行拼接
  70.                         treeMap.put(key, treeMap.get(key) + "_" + tuple2._1);
  71.                     } else {
  72.                         //当前map中不存在相同次数的广告id
  73.                         treeMap.put(key, tuple2._1);
  74.                     }
  75.                 }
  76.                 // 放入之后已经进行排序了  只需要找出前三即可
  77.                 ArrayList<Tuple2<String, Integer>> result = new ArrayList<>();
  78.                 // 如果总数据没有了  或者  结果够3个了  停止循环
  79.                 while (treeMap.size() > 0 && result.size() < 3) {
  80.                     //treeMap的lastEntry()方法会返回key值最大的键值对
  81.                     Map.Entry<Integer, String> entry = treeMap.lastEntry();
  82.                     //将次数相同的广告id遍历
  83.                     String[] ids = entry.getValue().split("_");
  84.                     for (String id : ids) {
  85.                         result.add(new Tuple2<>(id, entry.getKey()));
  86.                     }
  87.                     // 取出一个数据  用完之后删除
  88.                     treeMap.remove(entry.getKey());
  89.                 }
  90.                 return result;
  91.             }
  92.         });
  93.  
  94.         result. collect().forEach(System.out::println);
  95.  
  96.         //关闭sc
  97.         sc.stop();
  98.     }
  99. }

4.RDD序列化

在实际开发中我们往往需要自己定义一些对于RDD的操作，那么此时需要注意的是，初始化工作是在Driver端进行的，而实际运行程序是在Executor端进行的，这就涉及到了跨进程通信，是需要序列化的。

4.1 序列化异常

  1. package com.atguigu.serializable;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaRDD;
  5. import org.apache.spark.api.java.JavaSparkContext;
  6. import org.apache.spark.api.java.function.Function;
  7.  
  8. import java.util.Arrays;
  9.  
  10. public class Test01_Ser {
  11.     public static void main(String[] args) {
  12.         //创建spark配置对象
  13.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  14.         //创建sc环境
  15.         JavaSparkContext sc = new JavaSparkContext(conf);
  16.         //编写代码
  17.         //如果User不继承Serializable该代码会报错
  18.         //自定义的rdd初始化是在driver端进行的，
  19.         // 实际运行程序实在executor端进行，就涉及到网络通信，所以需要序列化
  20.         User zhangsan = new User("zhangsan", 19);
  21.         User lisi = new User("lisi", 18);
  22.  
  23.         JavaRDD<User> userRDD = sc.parallelize(Arrays.asList(zhangsan, lisi));
  24.  
  25.         JavaRDD<User> result = userRDD.map(new Function<User, User>() {
  26.             @Override
  27.             public User call(User v1) throws Exception {
  28.                 return new User(v1.getName(), v1.getAge() + 1);
  29.             }
  30.         });
  31.         result. collect().forEach(System.out::println);
  32.         //关闭sc
  33.         sc.stop();
  34.     }
  35. }

4.2Kryo序列化框架

Java的序列化能够序列化任何的类。但是比较重，序列化后对象的体积也比较大。

Spark出于性能的考虑，Spark2.0开始支持另外一种Kryo序列化机制。Kryo速度是Serializable的10倍。当RDD在Shuffle数据的时候，简单数据类型、数组和字符串类型已经在Spark内部使用Kryo来序列化。

5.RDD的依赖关系

5.1查看血缘关系

RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。

 WordCount的  .toDebugString  方法调用后的结果

 (2) ShuffledRDD[5] at reduceByKey at WordCount1.java:57 []
 +-(2) MapPartitionsRDD[3] at mapToPair at WordCount1.java:36 []
    |  MapPartitionsRDD[2] at flatMap at WordCount1.java:27 []
    |  input/2.txt MapPartitionsRDD[1] at textFile at WordCount1.java:24 []
    |  input/2.txt HadoopRDD[0] at textFile at WordCount1.java:24 []

 5.2窄依赖

窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition使用（一对一or多对一），窄依赖我们形象的比喻为独生子女。

5.3宽依赖

宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖（只能是一对多），会引起Shuffle，总结：宽依赖我们形象的比喻为超生。

具有宽依赖的transformations包括：sort、reduceByKey、groupByKey、join和调用rePartition函数的任何操作。

宽依赖对Spark去评估一个transformations有更加重要的影响，比如对性能的影响。

在不影响业务要求的情况下，要尽量避免使用有宽依赖的转换算子，因为有宽依赖，就一定会走shuffle，影响性能。

5.4Stage任务划分

1）DAG有向无环图

DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。例如，DAG记录了RDD的转换过程和任务的阶段。

 

 2）任务运行的整体流程

 

3）RDD任务切分中间分为：Application、Job、Stage和Task

（1）Application：初始化一个SparkContext即生成一个Application；

（2）Job：一个Action算子就会生成一个Job；

（3）Stage：Stage等于宽依赖的个数加1；

（4）Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。

注意：Application->Job->Stage->Task每一层都是1对n的关系

5.5RDD  Cache缓存

RDD通过Cache或者Persist方法将前面的计算结果缓存，默认情况下会把数据以序列化的形式缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的action算子时，该RDD将会被缓存在计算节点的内存中，并供后面重用。

Cache底层调用的就是Persist方法,缓存级别默认用的是MEMORY_ONLY

RDD.cache()

Persist方法可以更改存储级别，使用Persist方法有以下的参数，disk表示存储再磁盘，memory表示存储再内存中，最后的2表示两份数据

存储再内存中的数据，有可能会因为运行算子内存不够的时候，会被清掉，

两个方法都不会影响血缘关系

NONE

DISK_ONLY

DISK_ONLY_2

MEMORY_ONLY

 MEMORY_ONLY_2 

 MEMORY_ONLY_SER 

MEMORY_ONLY_SER_2 

 MEMORY_AND_DISK

MEMORY_AND_DISK_2

 MEMORY_AND_DISK_SER 

MEMORY_AND_DISK_SER_2 

OFF_HEAP

  1. package com.atguigu.cache;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.Function2;
  8. import org.apache.spark.api.java.function.PairFlatMapFunction;
  9. import org.apache.spark.storage.StorageLevel;
  10. import scala.Tuple2;
  11.  
  12. import java.util.ArrayList;
  13. import java.util.Iterator;
  14.  
  15. public class Test01_Cache {
  16.     public static void main(String[] args) throws InterruptedException {
  17.         //创建spark配置对象
  18.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  19.         //创建sc环境
  20.         JavaSparkContext sc = new JavaSparkContext(conf);
  21.         //编写代码
  22.         JavaRDD<String> lineRDD = sc.textFile("input/2.txt");
  23.  
  24.         JavaPairRDD<String, Integer> wordPairRDD = lineRDD.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
  25.             @Override
  26.             public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
  27.                 ArrayList<Tuple2<String, Integer>> tuple2s = new ArrayList<>();
  28.                 String[] words = s.split(" ");
  29.                 for (String word : words) {
  30.                     tuple2s.add(new Tuple2<>(word, 1));
  31.                 }
  32.                 return tuple2s.iterator();
  33.             }
  34.         });
  35. //        wordPairRDD.cache();
  36.         wordPairRDD.persist(StorageLevel.MEMORY_ONLY());
  37.  
  38.         wordPairRDD.persist(StorageLevel.MEMORY_ONLY());
  39.  
  40.         JavaPairRDD<String, Integer> resultRDD = wordPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
  41.             @Override
  42.             public Integer call(Integer v1, Integer v2) throws Exception {
  43.                 return v1 + v2;
  44.             }
  45.         });
  46.         resultRDD. collect().forEach(System.out::println);
  47.         System.out.println(resultRDD.toDebugString());
  48.         resultRDD. collect().forEach(System.out::println);
  49.        // Thread.sleep(120000);
  50.         //关闭sc
  51.         sc.stop();
  52.     }
  53. }

5.6 RDD   CheckPoint检查点

1）检查点：是通过将RDD中间结果写入磁盘。

2）为什么要做检查点？

由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。

3）检查点存储路径：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统

4）检查点数据存储格式为：二进制的文件

5）检查点切断血缘：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。

6）检查点触发时间：对RDD进行Checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍。

  1. package com.atguigu.cache;
  2.  
  3. import org.apache.spark.SparkConf;
  4. import org.apache.spark.api.java.JavaPairRDD;
  5. import org.apache.spark.api.java.JavaRDD;
  6. import org.apache.spark.api.java.JavaSparkContext;
  7. import org.apache.spark.api.java.function.PairFlatMapFunction;
  8. import scala.Tuple2;
  9.  
  10. import java.util.ArrayList;
  11. import java.util.Iterator;
  12.  
  13. public class Test02_CheckPoint {
  14.     public static void main(String[] args) throws InterruptedException {
  15.         //创建spark配置对象
  16.         SparkConf conf = new SparkConf().setAppName("core").setMaster("local[*]");
  17.         //创建sc环境
  18.         JavaSparkContext sc = new JavaSparkContext(conf);
  19.         sc.setCheckpointDir("ck");
  20.         //编写代码
  21.         JavaRDD<String> lineRDD = sc.textFile("input/2.txt");
  22.  
  23.         JavaPairRDD<String, Integer> wordPairRDD = lineRDD.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
  24.             @Override
  25.             public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
  26.                 ArrayList<Tuple2<String, Integer>> tuple2s = new ArrayList<>();
  27.                 String[] words = s.split(" ");
  28.                 for (String word : words) {
  29.                     tuple2s.add(new Tuple2<>(word, 1));
  30.                 }
  31.                 return tuple2s.iterator();
  32.             }
  33.         });
  34.         //加上cache()方法发后，就是3个job
  35.         wordPairRDD.cache();
  36.         //如果只使用checkpoint()方法的话，会有4个job
  37.         //因为checkpoint会自己再计算一边
  38.         wordPairRDD.checkpoint();
  39.  
  40.         wordPairRDD. collect().forEach(System.out::println);
  41.         wordPairRDD. collect().forEach(System.out::println);
  42.         wordPairRDD. collect().forEach(System.out::println);
  43.  
  44.         Thread.sleep(120000);
  45.  
  46.         //关闭sc
  47.         sc.stop();
  48.     }

5.7 键值对RDD数据分区

Spark目前支持Hash分区、Range分区和用户自定义分区。Hash分区为当前的默认分区。分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle后进入哪个分区和Reduce的个数。

1）注意：

（1）只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区的值是None

（2）每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/凡人多烦事01/article/detail/466659
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号