云计算与大数据实验五 MapReduce编程

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 云计算与大数据实验五 MapReduce编程

一、实验目的

  1. 了解Mapper类,Reducer类和Job类
  2. 掌握什么是MapReduce及使用MapReduce进行运算
  3. 掌握挖掘父子辈关系,给出祖孙辈关系的表格

二、实验内容

  1. 使用Map/Reduce计算班级中年龄最大的学生
  2. 使用Map/Reduce编程实现文件合并和去重操作
  3. 对给定的表格进行信息挖掘
  4. 编写实现日期操作的程序

三、实验步骤

(一)使用Map/Reduce计算班级中年龄最大的学生

什么是MapReduce

MapReduce是一种可用于数据处理的编程模型,我们现在设想一个场景,你接到一个任务,任务是:挖掘分析我国气象中心近年来的数据日志,该数据日志大小有3T,让你分析计算出每一年的最高气温,如果你现在只有一台计算机,如何处理呢?我想你应该会读取这些数据,并且将读取到的数据与目前的最大气温值进行比较。比较完所有的数据之后就可以得出最高气温了。不过以我们的经验都知道要处理这么多数据肯定是非常耗时的。

创建file01输入内容:

Hello World Bye World

创建file02输入内容:

Hello Hadoop Goodbye Hadoop

将文件上传到HDFS/usr/input/目录下:

不要忘了启动DFSstart-dfs.sh

然后创建文件夹并上传:

点击评测,运行代码,可以看到/usr/output目录下已经生成了文件。

我们来查看part--r-00000文件的内容:

可以看到统计的数据已经生成在文件中了。

编程要求

使用MapReduce计算班级每个学生的最好成绩,输入文件路径为/user/test/input,请将计算后的结果输出到/user/test/output/目录下。

输入文件的数据格式如下: 张三 12李四 13张三 89李四 92...

依照如上格式你应该输出:

张三 89李四 92

相关代码:

1. 1. //首先在命令行启动 hadoop: start-all.sh
2. 2. import java.io.IOException;
3. 3. import java.util.StringTokenizer;
4. 4. 
5. 5. import java.io.IOException;
6. 6. import java.util.StringTokenizer;
7. 7. import org.apache.hadoop.conf.Configuration;
8. 8. import org.apache.hadoop.fs.Path;
9. 9. import org.apache.hadoop.io.*;
10. 10. import org.apache.hadoop.io.Text;
11. 11. import org.apache.hadoop.mapreduce.Job;
12. 12. import org.apache.hadoop.mapreduce.Mapper;
13. 13. import org.apache.hadoop.mapreduce.Reducer;
14. 14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
15. 15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
16. 16. import org.apache.hadoop.util.GenericOptionsParser;
17. 17. 
18. 18. public class WordCount {
19. 19.     /********** Begin **********/
20. 20.     public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
21. 21.         private final static IntWritable one = new IntWritable(1);
22. 22.         private Text word = new Text();
23. 23.         private int maxValue = 0;
24. 24.         public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
25. 25.             StringTokenizer itr = new StringTokenizer(value.toString(),"\n");
26. 26.             while (itr.hasMoreTokens()) {
27. 27.                 String[] str = itr.nextToken().split(" ");
28. 28.                 String name = str[0];
29. 29.                 one.set(Integer.parseInt(str[1]));
30. 30.                 word.set(name);
31. 31.                 context.write(word,one);
32. 32.             }
33. 33.             //context.write(word,one);
34. 34.         }
35. 35.     }
36. 36.     public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
37. 37.         private IntWritable result = new IntWritable();
38. 38.         public void reduce(Text key, Iterable<IntWritable> values, Context context)
39. 39.                 throws IOException, InterruptedException {
40. 40.             int maxAge = 0;
41. 41.             int age = 0;
42. 42.             for (IntWritable intWritable : values) {
43. 43.                 maxAge = Math.max(maxAge, intWritable.get());
44. 44.             }
45. 45.             result.set(maxAge);
46. 46.             context.write(key, result);
47. 47.         }
48. 48.     }
49. 49.     public static void main(String[] args) throws Exception {
50. 50.         Configuration conf = new Configuration();
51. 51.         Job job = new Job(conf, "word count");
52. 52.         job.setJarByClass(WordCount.class);
53. 53.         job.setMapperClass(TokenizerMapper.class);
54. 54.         job.setCombinerClass(IntSumReducer.class);
55. 55.         job.setReducerClass(IntSumReducer.class);
56. 56.         job.setOutputKeyClass(Text.class);
57. 57.         job.setOutputValueClass(IntWritable.class);
58. 58.         String inputfile = "/user/test/input";
59. 59.         String outputFile = "/user/test/output/";
60. 60.         FileInputFormat.addInputPath(job, new Path(inputfile));
61. 61.         FileOutputFormat.setOutputPath(job, new Path(outputFile));
62. 62.         job.waitForCompletion(true);
63. 63.     /********** End **********/
64. 64.     }
65. 65. }

(二)使用Map/Reduce编程实现文件合并和去重操作

map类

首先我们来看看Mapper对象:

在编写MapReduce程序时,要编写一个类继承Mapper类,这个Mapper类是一个泛型类型,它有四个形参类型,分别指定了map()函数的输入键,输入值,和输出键,输出值的类型。就第一关的例子来说,输入键是一个长整型,输入值是一行文本,输出键是单词,输出值是单词出现的次数。

Hadoop提供了一套可优化网络序列化传输的基本类型,而不是直接使用Java内嵌的类型。这些类型都在org.apache.hadoop.io包中,这里使用LongWritable(相当于Java中的Long类型),Text类型(相当于Java中的String类型)和IntWritable(相当于Integer类型)。

map()函数的输入是一个键和一个值,我们一般首先将包含有一行输入的text值转换成JavaString类型,然后再使用对字符串操作的类或者其他方法进行操作即可。

Reducer类

同样Reducer也有四个参数类型用于指定输入和输出类型,reduce()函数的输入类型必须匹配map函数的输出类型,即Text类型和IntWritable类型,在这种情况下,reduce函数的输出类型也必须是TextIntWritable类型,即分别输出单词和次数。

Job类

一般我们用Job对象来运行MapReduce作业,Job对象用于指定作业执行规范,我们可以用它来控制整个作业的运行,我们在Hadoop集群上运行这个作业时,要把代码打包成一个JAR文件(Hadoop在集群上发布的这个文件),不用明确指定JAR文件的名称,在Job对象的setJarByClass()函数中传入一个类即可,Hadoop利用这个类来查找包含他的JAR文件。addInputPath()函数和setOutputPath()函数用来指定作业的输入路径和输出路径。值的注意的是,输出路径在执行程序之前不能存在,否则Hadoop会拒绝执行你的代码。

最后我们使用waitForCompletion()方法提交代码并等待执行,该方法唯一的参数是一个布尔类型的值,当该值为true时,作业会把执行过程打印到控制台,该方法也会返回一个布尔值,表示执行的成败。

编程要求

对于两个输入文件,即文件file1和文件file2,请编写MapReduce程序,对两个文件进行合并,并剔除其中重复的内容,得到一个新的输出文件file3。 为了完成文件合并去重的任务,你编写的程序要能将含有重复内容的不同文件合并到一个没有重复的整合文件,规则如下:

  • 第一列按学号排列;
  • 学号相同,按x,y,z排列;
  • 输入文件路径为:/user/tmp/input/
  • 输出路径为:/user/tmp/output/

程序会对你编写的代码进行测试: 输入已经指定了测试文本数据:需要你的程序输出合并去重后的结果。 下面是输入文件和输出文件的一个样例供参考。

输入文件file1的样例如下: 20150101 x20150102 y20150103 x20150104 y20150105 z20150106 x

输入文件file2的样例如下: 20150101 y20150102 y20150103 x20150104 z20150105 y

根据输入文件file1file2合并得到的输出文件file3的样例如下:

20150101 x20150101 y20150102 y20150103 x20150104 y20150104 z20150105 y20150105 z20150106 x

相关代码:

1. 1. import java.io.IOException;
2. 2. 
3. 3. 
4. 4. import java.util.*;
5. 5. import org.apache.hadoop.conf.Configuration;
6. 6. import org.apache.hadoop.fs.Path;
7. 7. import org.apache.hadoop.io.*;
8. 8. import org.apache.hadoop.mapreduce.Job;
9. 9. import org.apache.hadoop.mapreduce.Mapper;
10. 10. import org.apache.hadoop.mapreduce.Reducer;
11. 11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
12. 12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
13. 13. import org.apache.hadoop.util.GenericOptionsParser;
14. 14. public class Merge {
15. 15. 
16. 16.     /**
17. 17.      * @param args
18. 18.      * 对A,B两个文件进行合并,并剔除其中重复的内容,得到一个新的输出文件C
19. 19.      */
20. 20.     //在这重载map函数,直接将输入中的value复制到输出数据的key上 注意在map方法中要抛出异常:throws IOException,InterruptedException
21. 21.     /********** Begin **********/
22. 22.     public static class Map extends Mapper<LongWritable, Text, Text, Text >
23. 23.     {
24. 24.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
25. 25.                 throws IOException, InterruptedException {
26. 26.             String str = value.toString();
27. 27.             String[] data = str.split(" ");
28. 28.             Text t1= new Text(data[0]);
29. 29.             Text t2 = new Text(data[1]);
30. 30.             context.write(t1,t2);
31. 31.         }
32. 32.     } 
33. 33.     /********** End **********/
34. 34. 
35. 35.     //在这重载reduce函数,直接将输入中的key复制到输出数据的key上  注意在reduce方法上要抛出异常:throws IOException,InterruptedException
36. 36.     /********** Begin **********/
37. 37.     public static class Reduce  extends Reducer<Text, Text, Text, Text>
38. 38.     {
39. 39.         protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
40. 40.                 throws IOException, InterruptedException {
41. 41.             List<String> list = new ArrayList<>();
42. 42.             for (Text text : values) {
43. 43.                 String str = text.toString();
44. 44.                 if(!list.contains(str)){
45. 45.                     list.add(str);
46. 46.                 }
47. 47.             }
48. 48.             Collections.sort(list);
49. 49.             for (String text : list) {
50. 50.                 context.write(key, new Text(text));
51. 51.             }
52. 52.         }
53. 53.     /********** End **********/
54. 54.     }
55. 55. 
56. 56.     public static void main(String[] args) throws Exception{
57. 57.         Configuration conf = new Configuration();
58. 58.          Job job = new Job(conf, "word count");
59. 59.         job.setJarByClass(Merge.class);
60. 60.         job.setMapperClass(Map.class);
61. 61.         job.setCombinerClass(Reduce.class);
62. 62.         job.setReducerClass(Reduce.class);
63. 63.         job.setOutputKeyClass(Text.class);
64. 64.         job.setOutputValueClass(Text.class);
65. 65.         String inputPath = "/user/tmp/input/";  //在这里设置输入路径
66. 66.         String outputPath = "/user/tmp/output/";  //在这里设置输出路径
67. 67.         FileInputFormat.addInputPath(job, new Path(inputPath));
68. 68.         FileOutputFormat.setOutputPath(job, new Path(outputPath));
69. 69.         System.exit(job.waitForCompletion(true) ? 0 : 1);
70. 70.     }
71. 71. }

(三)对给定的表格进行信息挖掘

编程要求

你编写的程序要能挖掘父子辈关系,给出祖孙辈关系的表格。规则如下:

  • 孙子在前,祖父在后;
  • 输入文件路径:/user/reduce/input
  • 输出文件路径:/user/reduce/output

输入文件内容如下: child parentSteven LucySteven JackJone LucyJone JackLucy MaryLucy FrankJack AliceJack JesseDavid AliceDavid JessePhilip DavidPhilip AlmaMark DavidMark Alma

输出文件内容如下:

1. grand_child    grand_parent
2. Mark    Jesse
3. Mark    Alice
4. Philip    Jesse
5. Philip    Alice
6. Jone    Jesse
7. Jone    Alice
8. Steven    Jesse
9. Steven    Alice
10. Steven    Frank
11. Steven    Mary
12. Jone    Frank
13. Jone    Mary

相关代码:

1. 1. import java.io.IOException;
2. 2. import java.util.*;
3. 3. 
4. 4. import org.apache.hadoop.conf.Configuration;
5. 5. import org.apache.hadoop.fs.Path;
6. 6. import org.apache.hadoop.io.IntWritable;
7. 7. import org.apache.hadoop.io.Text;
8. 8. import org.apache.hadoop.mapreduce.Job;
9. 9. import org.apache.hadoop.mapreduce.Mapper;
10. 10. import org.apache.hadoop.mapreduce.Reducer;
11. 11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
12. 12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
13. 13. import org.apache.hadoop.util.GenericOptionsParser;
14. 14. 
15. 15. public class simple_data_mining {
16. 16.     public static int time = 0;
17. 17. 
18. 18.     /**
19. 19.      * @param args
20. 20.      * 输入一个child-parent的表格
21. 21.      * 输出一个体现grandchild-grandparent关系的表格
22. 22.      */
23. 23.     //Map将输入文件按照空格分割成child和parent,然后正序输出一次作为右表,反序输出一次作为左表,需要注意的是在输出的value中必须加上左右表区别标志
24. 24.     public static class Map extends Mapper<Object, Text, Text, Text>{
25. 25.         public void map(Object key, Text value, Context context) throws IOException,InterruptedException{
26. 26.             /********** Begin **********/
27. 27. 
28. 28. 
29. 29.             String child_name = new String();
30. 30.             String parent_name = new String();
31. 31.             String relation_type = new String();
32. 32.             String line = value.toString();
33. 33.             int i = 0;
34. 34.             while(line.charAt(i) != ' '){
35. 35.                 i++;
36. 36.             }
37. 37.             String[] values = {line.substring(0,i),line.substring(i+1)};
38. 38.             if(values[0].compareTo("child") != 0){
39. 39.                 child_name = values[0];
40. 40.                 parent_name = values[1];
41. 41.                 relation_type = "1";//左右表区分标志
42. 42.                 context.write(new Text(values[1]), new Text(relation_type+"+"+child_name+"+"+parent_name));
43. 43.                 //左表
44. 44.                 relation_type = "2";
45. 45.                 context.write(new Text(values[0]), new Text(relation_type+"+"+child_name+"+"+parent_name));
46. 46.                 //右表
47. 47.             }
48. 48. 
49. 49. 
50. 50. 
51. 51. 
52. 52. 
53. 53. 
54. 54.             /********** End **********/
55. 55.         }
56. 56.     }
57. 57. 
58. 58.     public static class Reduce extends Reducer<Text, Text, Text, Text>{
59. 59.         public void reduce(Text key, Iterable<Text> values,Context context) throws IOException,InterruptedException{
60. 60.                 /********** Begin **********/
61. 61.             if(time == 0){   //输出表头
62. 62.                 context.write(new Text("grand_child"), new Text("grand_parent"));
63. 63.                 time++;
64. 64.             }
65. 65.             int grand_child_num = 0;
66. 66.             String grand_child[] = new String[10];
67. 67.             int grand_parent_num = 0;
68. 68.             String grand_parent[]= new String[10];
69. 69.             Iterator ite = values.iterator();
70. 70.             while(ite.hasNext()){
71. 71.                 String record = ite.next().toString();
72. 72.                 int len = record.length();
73. 73.                 int i = 2;
74. 74.                 if(len == 0) continue;
75. 75.                 char relation_type = record.charAt(0);
76. 76.                 String child_name = new String();
77. 77.                 String parent_name = new String();
78. 78.                 //获取value-list中value的child
79. 79.                 while(record.charAt(i) != '+'){
80. 80.                     child_name = child_name + record.charAt(i);
81. 81.                     i++;
82. 82.                 }
83. 83.                 i=i+1;
84. 84.                 //获取value-list中value的parent
85. 85.                 while(i<len){
86. 86.                     parent_name = parent_name+record.charAt(i);
87. 87.                     i++;
88. 88.                 }
89. 89.                 //左表,取出child放入grand_child
90. 90.                 if(relation_type == '1'){
91. 91.                     grand_child[grand_child_num] = child_name;
92. 92.                     grand_child_num++;
93. 93.                 }
94. 94.                 else{//右表,取出parent放入grand_parent
95. 95.                     grand_parent[grand_parent_num] = parent_name;
96. 96.                     grand_parent_num++;
97. 97.                 }
98. 98.             }
99. 99.             if(grand_parent_num != 0 && grand_child_num != 0 ){
100. 100.                 for(int m = 0;m<grand_child_num;m++){
101. 101.                     for(int n=0;n<grand_parent_num;n++){
102. 102.                         context.write(new Text(grand_child[m]), new Text(grand_parent[n]));
103. 103.                         //输出结果
104. 104.                     }
105. 105.                 }
106. 106.             }
107. 107. 
108. 108.                 /********** End **********/
109. 109. 
110. 110.         }
111. 111.     }
112. 112.     public static void main(String[] args) throws Exception{
113. 113.         // TODO Auto-generated method stub
114. 114.         Configuration conf = new Configuration();
115. 115.         Job job = Job.getInstance(conf,"Single table join");
116. 116.         job.setJarByClass(simple_data_mining.class);
117. 117.         job.setMapperClass(Map.class);
118. 118.         job.setReducerClass(Reduce.class);
119. 119.         job.setOutputKeyClass(Text.class);
120. 120.         job.setOutputValueClass(Text.class);
121. 121.         String inputPath = "/user/reduce/input";   //设置输入路径
122. 122.         String outputPath = "/user/reduce/output";   //设置输出路径
123. 123.         FileInputFormat.addInputPath(job, new Path(inputPath));
124. 124.         FileOutputFormat.setOutputPath(job, new Path(outputPath));
125. 125.         System.exit(job.waitForCompletion(true) ? 0 : 1);
126. 126. 
127. 127.     }
128. 128. 
129. 129. }

四、实验心得

掌握了什么是MapReduce及使用MapReduce进行运算

掌握了挖掘父子辈关系,给出祖孙辈关系的表格

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
5月前
|
存储 大数据 云计算
大数据与云计算
大数据与云计算
147 2
|
2月前
|
分布式计算 大数据 Hadoop
揭秘MapReduce背后的魔法:从基础类型到高级格式,带你深入理解这一大数据处理利器的奥秘与实战技巧,让你从此不再是编程门外汉!
【8月更文挑战第17天】MapReduce作为分布式计算模型,是大数据处理的基石。它通过Map和Reduce函数处理大规模数据集,简化编程模型,使开发者聚焦业务逻辑。MapReduce分单阶段和多阶段,支持多种输入输出格式如`TextInputFormat`和`SequenceFileInputFormat`。例如,简单的单词计数程序利用`TextInputFormat`读取文本行并计数;而`SequenceFileInputFormat`适用于高效处理二进制序列文件。合理选择类型和格式可有效解决大数据问题。
38 1
|
2月前
|
分布式计算 并行计算 大数据
【数据挖掘】百度2015大数据云计算研发笔试卷
百度2015年大数据云计算研发笔试卷的题目总结,涵盖了Hadoop、Spark、MPI计算框架特点、TCP连接建立过程、数组最大和问题、二分查找实现以及灯泡开关问题,提供了部分题目的解析和伪代码。
49 1
|
2月前
|
安全 Java 网络安全
云计算时代下的网络安全挑战与应对策略Java编程中的异常处理:从基础到高级
在云服务不断深入各行各业的今天,网络安全问题也随之凸显。本文将探讨云计算环境下的安全风险,并提出相应的防护措施,以期为相关行业提供参考和指导。 在Java的世界里,异常处理是代码健壮性的守护神。它不仅保护程序免于意外崩溃,还提供了一种优雅的方式来响应错误。本文将带你领略异常处理的艺术,从简单的try-catch语句到复杂的自定义异常和finally块的神秘力量,我们将一起探索如何让Java程序在面对不确定性时,依然能够优雅地起舞。
|
4月前
|
分布式计算 Hadoop Java
MapReduce编程模型——在idea里面邂逅CDH MapReduce
MapReduce编程模型——在idea里面邂逅CDH MapReduce
75 15
|
3月前
|
网络协议 Unix 数据安全/隐私保护
云计算存储问题之在编程场景中接口如何解决
云计算存储问题之在编程场景中接口如何解决
|
4月前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
48 1
|
4月前
|
机器学习/深度学习 存储 算法
云计算和大数据处理
云计算和大数据处理
65 1
|
4月前
|
存储 分布式计算 Hadoop
MapReduce编程模型——自定义序列化类实现多指标统计
MapReduce编程模型——自定义序列化类实现多指标统计
34 0
|
4月前
|
机器学习/深度学习 分布式计算 并行计算
MapReduce是一种用于并行计算的编程模型和处理大规模数据集的实现
MapReduce是一种用于并行计算的编程模型和处理大规模数据集的实现
55 0