devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spark Java实战

作者：笔触狂放9 | 2024-04-22 03:55:51

赞

踩

spark java

目录

一、数据去重

1.1实例描述

1.2 解题思路

1.3 代码展示

1.4 运行结果展示

 二、数据排序

2.1 实例描述

2.2 解题思路

2.3 代码展示

2.4 运行结果展示

 三、平均成绩

3.1 实例描述

3.2 解题思路

3.3 代码展示

3.4 运行结果展示

 四、单表关联

4.1 实例描述

4.2 解题思路

4.3代码展示

4.4 运行结果展示

一、数据去重

 "数据去重"主要是为了掌握和利用并行化思想来对数据进行有意义的筛选。统计大数据集上的数据种类个数、从网站日志中计算访问地等这些看似庞杂的任务都会涉及数据去重。

1.1实例描述

对数据文件中的数据进行去重。数据文件中的每行都是一个数据。样例输入如下所示：

1）file1：

2012-3-1 a

2012-3-2 b

2012-3-3 c

2012-3-4 d

2012-3-5 a

2012-3-6 b

2012-3-7 c

2012-3-3 c

2）file2：

2012-3-1 b

2012-3-2 a

2012-3-3 b

2012-3-4 d

2012-3-5 a

2012-3-6 c

2012-3-7 d

2012-3-3 c

样例输出如下所示：

2012-3-1 a

2012-3-1 b

2012-3-2 a

2012-3-2 b

2012-3-3 b

2012-3-3 c

2012-3-4 d

2012-3-5 a

2012-3-6 b

2012-3-6 c

2012-3-7 c

1.2 解题思路

 1.3 代码展示

  1. public class Deduplicate {
  2.     public static void main(String[] args){
  3.         //todo:1、构建sparkconf,设置配置信息
  4.         SparkConf conf = new SparkConf().setAppName("Deduplicated Application").setMaster("local[2]");;
  5.         //todo:2、构建java版的sparkContext
  6.         JavaSparkContext sc = new JavaSparkContext(conf);
  7.         //todo:3、读取数据文件
  8.         JavaRDD<String> dataRDD1 = sc.textFile("data/data1.txt");
  9.         JavaRDD<String> dataRDD2 = sc.textFile("data/data2.txt");
  10.         //todo:4、合并数据集
  11.         JavaRDD<String> dataRDD = dataRDD1.union(dataRDD2);
  12.         //todo:5、生成<值、" ">键值对
  13.         JavaPairRDD<String,String> map =dataRDD.mapToPair(
  14.                 new PairFunction<String, String, String>() {
  15.                     public Tuple2<String, String> call(String s) throws Exception {
  16.                         return new Tuple2<String,String >(s,"");
  17.                         //获取数值作为key,空串作为value
  18.                     }
  19.                 }
  20.         );
  21.         //todo:6、通过合并键去重
  22.         JavaPairRDD<String,String> result = map.reduceByKey(
  23.                 new Function2<String, String, String>() {
  24.                     @Override
  25.                     public String call(String s, String s2) throws Exception {
  26.                         return s2;
  27.                     }
  28.                 }
  29.         );
  30.         //todo:7、输出
  31.        Object[] array = result.collect().toArray();
  32.         for (Object o : array) {
  33.             Tuple2<String,String> tuple2 = (Tuple2<String,String>) o;
  34.             System.out.println(tuple2._1);
  35.         }
  36.     }
  37. }

 1.4 运行结果展示

 

 二、数据排序

"数据排序"是许多实际任务执行时要完成的第一项工作，比如学生成绩评比、数据建立索引等。这个实例和数据去重类似，都是先对原始数据进行初步处理，为进一步的数据操作打好基础。下面进入这个示例。

2.1 实例描述

对输入文件中数据进行排序。输入文件中的每行内容均为一个数字，即一个数据。要求在输出中每行有两个间隔的数字，其中，第一个代表原始数据在原始数据集中的位次，第二个代表原始数据。

样例输入：

1）file1：

2

32

654

32

15

756

65223

2）file2：

5956

22

650

92

3）file3：  

26

54

6

样例输出：

1    2

2    6

3    15

4    22

5    26

6    32

7    32

8    54

9    92

10    650

11    654

12    756

13    5956

14    65223

2.2 解题思路

 2.3 代码展示

  1. /**
  2.  * @Author: xuliushen
  3.  * @Description:
  4.  * @Date Created in 2021-09-07 16:00
  5.  * @Modified by :
  6.  */
  7. public class Rank {
  8.     public static void main(String[] args){
  9.         //todo:1、构建sparkconf,设置配置信息
  10.         SparkConf conf = new SparkConf().setAppName("DataRank Application").setMaster("local[2]");;
  11.         //todo:2、构建java版的sparkContext
  12.         JavaSparkContext sc = new JavaSparkContext(conf);
  13.         //todo:3、读取数据文件
  14.         JavaRDD<String> dataRDD1 = sc.textFile("data/file1.txt");
  15.         JavaRDD<String> dataRDD2 = sc.textFile("data/file2.txt");
  16.         JavaRDD<String> dataRDD3 = sc.textFile("data/file3.txt");
  17.         //todo:4、合并数据集
  18.         JavaRDD<String> dataRDD = dataRDD1.union(dataRDD2).union(dataRDD3);
  19.         //todo:5、生成<值、" ">键值对
  20.         JavaPairRDD<Integer,String> map =dataRDD.mapToPair(
  21.                 new PairFunction<String, Integer, String>() {
  22.                     public Tuple2<Integer, String> call(String s) throws Exception {
  23.                         return new Tuple2<Integer,String >(Integer.parseInt(s)," ");
  24.                         //获取数值作为key,空串作为value
  25.                     }
  26.                 }
  27.         );
  28.         //todo:6、进行排序
  29.         JavaPairRDD<Integer,String> result = map.sortByKey();
  30.         //todo:7、转换成Array并输出
  31.         Object[] array = result.collect().toArray();
  32.         for (int i = 0; i < array.length; i++) {
  33.             Tuple2<Integer,String> tuple2 = (Tuple2<Integer,String>) array[i];
  34.             System.out.println(i+1+" "+ tuple2._1);
  35.         }
  36.     }
  37. }

 2.4 运行结果展示

 三、平均成绩

 "平均成绩"主要目的还是在重温经典"WordCount"例子，可以说是在基础上的微变化版，该实例主要就是实现一个计算学生平均成绩的例子。

3.1 实例描述

对输入文件中数据进行就算学生平均成绩。输入文件中的每行内容均为一个学生的姓名和他相应的成绩，如果有多门学科，则每门学科为一个文件。要求在输出中每行有两个间隔的数据，其中，第一个代表学生的姓名，第二个代表其平均成绩。

样本输入：

1）math：

张三    88

李四    99

王五    66

赵六    77

2）china：

 张三    78

李四    89

王五    96

赵六    67

3）english：

张三    80

李四    82

王五    84

赵六    86

样本输出：

张三    82

李四    90

王五    82

赵六    76 

3.2 解题思路

 3.3 代码展示

  1. /**
  2.  * @Author: xuliushen
  3.  * @Description:
  4.  * @Date Created in 2021-09-07 14:39
  5.  * @Modified by :
  6.  */
  7. public class Average {
  8.     public static void main(String args[]) {
  9.         //todo:1、构建sparkconf,设置配置信息
  10.         SparkConf conf = new SparkConf().setAppName("Average Application").setMaster("local[2]");;
  11.         //todo:2、构建java版的sparkContext
  12.         JavaSparkContext sc = new JavaSparkContext(conf);
  13.         //todo:3、读取数据文件
  14.         JavaRDD<String> dataRDD1 = sc.textFile("data/math.txt");
  15.         JavaRDD<String> dataRDD2 = sc.textFile("data/china.txt");
  16.         JavaRDD<String> dataRDD3 = sc.textFile("data/english.txt");
  17.         //todo:4、合并数据集
  18.         JavaRDD<String> lines = dataRDD1.union(dataRDD2).union(dataRDD3);
  19.         //todo:5、每一行转化为键值对
  20.         JavaPairRDD<String,Integer> map =lines.mapToPair(
  21.                 new PairFunction<String, String, Integer>() {
  22.                     public Tuple2<String, Integer> call(String s) throws Exception {
  23.                         String[] line = s.split(" ");
  24.                         return new Tuple2<String,Integer >(line[0],Integer.parseInt(line[1]));
  25.                         //获取第一个单词作为key,第二个转化为Long作为value
  26.                     }
  27.                 }
  28.         );
  29.  
  30.         JavaPairRDD<String, Tuple2<Integer, Integer>> data3 = map.mapValues(
  31.                 x -> new Tuple2<Integer, Integer>(x, 1));
  32.  
  33.         JavaPairRDD<String, Tuple2<Integer, Integer>> data4 = data3.reduceByKey((x, y) -> {
  34.             return new Tuple2<Integer, Integer>(x._1 + y._1, x._2 + y._2);
  35.         });
  36.  
  37.         JavaPairRDD<String, Integer> data5 = data4.mapToPair(x -> {
  38.             return new Tuple2<String, Integer>(x._1, x._2._1  / x._2._2);
  39.         });
  40.  
  41.         //todo:6、输出数据
  42.         System.out.println(data5.collect());
  43.     }
  44. }

 3.4 运行结果展示

 

 四、单表关联

前面的实例都是在数据上进行一些简单的处理，为进一步的操作打基础。"单表关联"这个实例要求从给出的数据中寻找所关心的数据，它是对原始数据所包含信息的挖掘。下面进入这个实例。

4.1 实例描述

实例中给出child-parent（孩子—父母）表，要求输出grandchild-grandparent（孙子—爷奶）表。样例输入如下所示。

file：

child        parent

Tom        Lucy

Tom        Jack

Jone        Lucy

Jone        Jack

Lucy        Mary

Lucy        Ben

Jack        Alice

Jack        Jesse

Terry        Alice

Terry        Jesse

Philip        Terry

Philip        Alma

Mark        Terry

Mark        Alma

家族树状关系谱：

图4.2-1 家族谱

样例输出如下所示。

file：

grandchild        grandparent

Tom            Alice

Tom            Jesse

Jone            Alice

Jone            Jesse

Tom            Mary

Tom            Ben

Jone            Mary

Jone            Ben

Philip             Alice

Philip            Jesse

Mark            Alice

Mark            Jesse 

 4.2 解题思路

 4.3代码展示

  1. /**
  2.  * @Author: xuliushen
  3.  * @Description:
  4.  * @Date Created in 2021-09-07 20:46
  5.  * @Modified by :
  6.  */
  7. public class SingleRelation {
  8.     public static void main(String[] args){
  9.         //todo:1、构建sparkconf,设置配置信息
  10.         SparkConf conf = new SparkConf().setAppName("SingleRelation Application").setMaster("local[2]");;
  11.         //todo:2、构建java版的sparkContext
  12.         JavaSparkContext sc = new JavaSparkContext(conf);
  13.         //todo:3、读取数据文件
  14.         JavaRDD<String> dataRDD= sc.textFile("data/relation.txt");
  15.         //去掉第一行
  16.         Object[] array = dataRDD.collect().toArray();
  17.         List list = Arrays.asList(array);
  18.         list = list.subList(1,list.size());//去掉
  19.         dataRDD = sc.parallelize(list);
  20.         //todo:4、把每一行映射成为键值对
  21.         //键值对<child,parent>
  22.         JavaPairRDD<String,String> map1 =dataRDD.mapToPair(
  23.                 new PairFunction<String, String, String>() {
  24.                     public Tuple2<String, String> call(String s) throws Exception {
  25.                         String[] line = s.split(" ");
  26.                         return new Tuple2<String,String>(line[0],"0"+line[1]);//0表示为父亲
  27.                         //获取第一个单词作为key,第二个转化为Long作为value
  28.                     }
  29.                 }
  30.         );
  31.         //键值对<parent,child>
  32.         JavaPairRDD<String,String> map2 =dataRDD.mapToPair(
  33.                 new PairFunction<String, String, String>() {
  34.                     public Tuple2<String, String> call(String s) throws Exception {
  35.                         String[] line = s.split(" ");
  36.                         return new Tuple2<String,String>(line[1],"1"+line[0]);//1表示为孩子
  37.                         //获取第一个单词作为key,第二个转化为Long作为value
  38.                     }
  39.                 }
  40.         );
  41.         //todo:5、合并
  42.         JavaPairRDD<String,String> map = map1.union(map2);
  43.         //todo:6、根据键建立祖孙联系
  44.         JavaPairRDD<String,String> data1 = map.reduceByKey(
  45.                 new Function2<String, String, String>() {
  46.                     @Override
  47.                     public String call(String s, String s2) throws Exception {
  48.                         return s+" "+s2;
  49.                     }
  50.                 }
  51.         );
  52.  
  53.         //todo:7、输出
  54.         System.out.println("grandchild grandparent");
  55.         Object[] data2 = data1.collect().toArray();
  56.         for (Object o : data2) {
  57.             String relation =((Tuple2<String,String>)o)._2;//取出关系字符串
  58.             String[] relations = relation.split(" ");
  59.             List<String> fathers = new ArrayList<>();
  60.             List<String> childs = new ArrayList<>();
  61.             int i = 0;
  62.             for (String s : relations) {
  63.                 if(s.charAt(0) == '0'){//父亲
  64.                     fathers.add(s.substring(1));
  65.                 }else{//孩子
  66.                     childs.add(s.substring(1));
  67.                 }
  68.             }
  69.             if(fathers.size()!=0 && childs.size()!=0){
  70.                 for (String child : childs) {
  71.                     for (String father : fathers) {
  72.                         System.out.println(child+" "+father);
  73.                     }
  74.                 }
  75.             }
  76.         }
  77.  
  78.     }
  79. }

 4.4 运行结果展示

 

 

 

 

 

 

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

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

  

闽ICP备14008679号