初识Hadoop-MapReduce

Posted 2021-01-31 voguesuper

原文地址

https://blog.csdn.net/zhruixuan/article/details/85549618

Hadoop的核心是HDFS，YARN，MapReduce。今天先来认识一下MapReduce。

MapReduce是什么

MapReduce是Hadoop中的一种处理大规模数据的编程模型，得益于MapReduce的并行计算，可以高效的处理大规模数据（一般是HDFS中存储的数据）。 顾名思义，MapReduce分为两个处理阶段（对于开发者来说），Map阶段和Reduce阶段。每个阶段都以Key-Value作为输入输出，Key-Value的类型由开发者选择。map阶段一般可以用于对原始数据进行预处理，过滤，数据校验等操作。数据经过map之后会经过MapReduce框架的Shuffle阶段，对map处理过的数据进行排序和分组。最后数据被传递给Reduce阶段进行结果数据的计算。

一个java版本的示例程序

我们的目标是从记录有每一年份每一个站点的气候信息的文件中找出每一年的最高温度。文件中每一行数据格式如下图，加粗表示的是年份和温度信息。

要实现MapReduce需要实现两个函数，map函数reduce函数。实现map函数可以通过实现Mapper借口来完成。

map函数

1. import java.io.IOException;
2. import org.apache.hadoop.io.IntWritable;
3. import org.apache.hadoop.io.LongtWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Mapper;
6. //Mapper的泛型分别表示map函数的输入键类型，输入值类型，输出键类型，输出值类型
7. public class MaxTemeratureMapper extents MapReduceBase implements Mapper<LongWritable,Text,Text,IntWritable>{
8. private static final int MISSING = 9999;
9. @Override
10. //输入键是一行文本的行数偏移值，输入值是一行文本
11. public void map(LongWritable key, Text value, Context context) throws IOException,InterrupedException{
12. String line = value.toString();
13. String year = line.substring(15,19);
14. int airTemperature;
15. if(line.charAt(87) == ‘+‘){
16. airTemperature = Integer.parseInt(line.substring(88,92));
17. }else{
18. airTemperature = Integer.parseInt(line.substring(87,92));
19. }
20. String quality = line.substring(92,93);
21. if(airTemperature != MISSING && quality.matches("[01456]")){
22. //输出给shuffle
23. context.write(new Text(year),new IntWritable(airTemperature));
24. }
25. }
26. }

类似的书写reducer函数 reducer函数的输入类型必须与map函数的输出类型一致。

reducer函数

1. import org.apache.hadoop.io.IntWritable;
2. import org.apache.hadoop.io.Text;
3. import org.apache.hadoop.mapreduce.Reducer;
4. import java.io.IOException;
5. public class MaxTemperatureReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
6. @Override
7. //经过shuffle分组处理后values变成了一个集合
8. public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException,InterrupedException{
9. int maxValue = Integer.MIN_VALUE;
10. for(IntWritable value : values){
11. maxValue = Math.max(maxValue,value.get());
12. }
13. context.write(key, new IntWritable(maxValue));
14. }
15. }

通过一个入口函数来启动MapReduce 入口main

1. import java.io.IOException;
2. import org.apache.hadoop.fs.Path;
3. import org.apache.hadoop.io.IntWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Job;
6. import org.apache.hadoop.mapreduce.input.FileOutputFormat;
7. import org.apache.hadoop.mapreduce.input.FileInputFormat;
8. public class MaxTemperature(){
9. public static void main(String[] args) throws Exception{
10. if(args.length != 2){
11. System.out.println("请输入文件路径和结果输出路径");
12. System.exit(-1);
13. }
14. Job job = new Job();
15. job.setJarByClass(MaxTemperature.class);
16. job.setJobName("max temperature");
17. //可以是单个文件，也可以是一个目录
18. //如果是目录，会将该目录下的所有文件当做输入
19. //可以多次调用该方法
20. FileInputFormat.addInputPath(job,new Path(args[0]));
21. //只能有一个输出路径
22. //在运行作业前该目录不应该存在，否则会报错
23. FileInputFormat.setOutputPath(job,new Path(args[1]));
24. job.setMapperClass(MaxTemeratureMapper.class);
25. job.setReducerClass(MaxTemperatureReducer.class);
26. job.setOutputKeyClass(Text.class);
27. job.setOutputValueClass(IntWritable.class);
28. //true表示将进度信息写到控制台
29. System.exit(job.waitForCompletion(true) ? 0 : 1);
30. }
31. }

setOutputKeyClass()和setOutputValueClass()方法控制reduce函数的输出类型，并且必须和Reduce类产生的相匹配。map函数的输出类型如果和reduce的输入类型不同，则必须通过setMapOutputKeyClass()和setMapOutputValueClass()方法类设置map函数的输出类型。 单机测试运行 //将应用添加到类路径 % export HADOOP_CLASSPATH=hadoop-example.jar //运行MaxTemperature % hadoop MaxTemperature input/ncdc/sample.txt output

在output目录可以看到输出文件 % cat output/part-r-00000 1949 111 1950 22

横向扩展（并行计算）

以上MapReduce可以通过YARN（资源管理系统）处理HDFS中的数据。这样Hadoop会将MapReduce计算转移到存储部分数据的各台机器上并行执行任务。

我们把客户端提交的MapReduce程序，数据，配置信息称为MapReduce作业(Job)。Hadoop将作业分成若干个任务(Task)来执行，其中包括两类任务map任务和reduce任务，这些任务运行在集群的节点上，通过YARN进行调度，如果一个任务失败，会在另一个节点上重新运行。 Hadoop将map的输入数据划分成大小相等的数据块，称为输入分片（一般为一个HDFS块大小，默认是128M），Hadoop为每个分片构建一个map任务，该任务运行用户的map函数。 Hadoop在存储有输入数据的节点（HDFS）上运行map任务可以提升效率（称为数据本地化优化），但是当存储有输入数据的节点有其他map任务时，Hadoop会尽量从某一数据块所在的机架中的一个节点上选择一个空闲的map槽（slot）来运行该map任务。 这也是为什么最佳分片的大小应该和HDFS的块大小相同，因为如果分片跨越两个数据块，那么对于任何一个HDFS节点都不太可能存储 这两个数据块。 map任务的输出只是会存到本地磁盘，而不会写入HDFS，因为是中间结果。

reduce任务不具备数据本地化优势，因为单个reduce的输入通常来自于所有的map任务输出，本例中我们只有一个reduce任务，他的输入就是所有的map任务输出。 reduce任务的数量不是由输入数据的大小决定的，而是需要独立制定。

单个Reduce任务数据流

多个Reduce任务数据流

reduce任务的输入通常来自不同的map任务，通过shuffle（混洗）分配到不同的reduce任务。

没有Reduce任务的数据流

没有Reduce任务时，数据处理可以完全并行，直接由map任务进行最终结果的HDFS写入。

Combiner函数减少数据传输

通过combiner函数可以减少map函数与reduce函数之间的数据传输。 可以为map制定一个combiner，combiner的输出作为reduce的输入。 举个例子 假设1950年的输入数据由两个map任务处理（因为他们在不同的分片中），假设第一个map输出如下： （1950,0） （1950,10） （1950,20） 第二个map输出如下： （1950,25） （1950,15） 那么reduce 函数调用是输入将是 （1950， [0，10,20,25,15]） 然后计算出 (1950，25) 其实我们可以对map的结果调用reduce计算出每个map的最大气温，然后再传递给reduce

1. import java.io.IOException;
2. import org.apache.hadoop.fs.Path;
3. import org.apache.hadoop.io.IntWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Job;
6. import org.apache.hadoop.mapreduce.input.FileOutputFormat;
7. import org.apache.hadoop.mapreduce.input.FileInputFormat;
8. public class MaxTemperature(){
9. public static void main(String[] args) throws Exception{
10. if(args.length != 2){
11. System.out.println("请输入文件路径和结果输出路径");
12. System.exit(-1);
13. }
14. Job job = new Job();
15. job.setJarByClass(MaxTemperature.class);
16. job.setJobName("max temperature");
17. //可以是单个文件，也可以是一个目录
18. //如果是目录，会将该目录下的所有文件当做输入
19. //可以多次调用该方法
20. FileInputFormat.addInputPath(job,new Path(args[0]));
21. //只能有一个输出路径
22. //在运行作业前该目录不应该存在，否则会报错
23. FileInputFormat.setOutputPath(job,new Path(args[1]));
24. job.setMapperClass(MaxTemeratureMapper.class);
25. //**设置Combiner**
26. job.setCombinerClass(MaxTemperatureReducer.class);
27. job.setReducerClass(MaxTemperatureReducer.class);
28. job.setOutputKeyClass(Text.class);
29. job.setOutputValueClass(IntWritable.class);
30. //true表示将进度信息写到控制台
31. System.exit(job.waitForCompletion(true) ? 0 : 1);
32. }
33. }