005_hadoop中MapReduce详解_2

前面介绍了的MapReduce的入门。利用了一个hadoop自带的例子来说明MapReduce的流程。现在我们自己动手写一个小例子来锻炼一下。

问题描述：现在有一个文件，文件内容如下：

黄晓明 89

刘杰 48

黄晓明 78

郑爽 90

……

求学生的平均成绩？

分析：

1.在Map阶段我们的输入可以每行读取，生成类似<行号,行内容>即：

<1,黄晓明 89>

<2,刘杰 48>

<3,黄晓明 78>

等等。

2.然后这些记录进入Map函数。我们要充分利用Map--->洗牌--->Reduce这中间的洗牌操作。将Key相同的放在一起。Value变成一个List

3.构建Map的输出（即Reduce的输入）

<黄晓明,89>

<刘杰,48>

<黄晓明,78>

等等。

4.遍历相同Key的Value值，进行累加，求得总分数，然后除以科目总数，这样就得到最后结果

代码实现：

//继承Mapper，重新Map方法
	public static class MyMap extends Mapper<LongWritable, Text, Text, IntWritable>{
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			System.out.println(line);
			String [] arr = line.split(" ");
			Text name = new Text(arr[0]);
			String score = arr[1];
			IntWritable s = new IntWritable(Integer.parseInt(score));
			context.write(name, s);
		}
	}

Mapper处理的数据是由InputFormat分解过的数据集，其中InputFormat的作用是将数据集切割成小的InputSplit，每个InputSplit都由一个Map去处理。

//继承Reducer，重写reduce方法
	public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
		@Override
		protected void reduce(Text arg0, Iterable<IntWritable> arg1, Context arg2) throws IOException, InterruptedException {
			int sum = 0 ;
			int count = 0 ;
			Iterator<IntWritable> it = arg1.iterator();
			while(it.hasNext()){
				sum += it.next().get();
				count ++;
			}
			int ave = (int)sum/count;
			arg2.write(arg0, new IntWritable(ave));
		}
	}

 测试程序

public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
	    
	    Job job = new Job(conf, "average");
	    job.setJarByClass(TestMR.class);
	    job.setMapperClass(MyMap.class);
	    job.setCombinerClass(MyReduce.class);
	    job.setReducerClass(MyReduce.class);
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(IntWritable.class);
	    FileInputFormat.addInputPath(job, new Path("/user/text.txt")); //Map的输入
	    FileOutputFormat.setOutputPath(job, new Path("/user/helloMR/success"));//Reduce的输出
	    System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

 源码我已经上传了

是不是非常简单的一个例子

在开发上面代码时，分别覆盖了Mapper里面的map方法和Reducer里面的reduce方法。我们还发现有其他方法可以覆盖

 

 既然看到了，就简单说一下其他的三个方法：

1.setup：setup函数是在task启动开始就调用

2.cleanup：cleanup函数是在task销毁之前执行

3.run：run函数是Map类或者Reduce类的启动方法：先调用setup函数，然后针对每个key调用一次Map或者Reduce，最后在销毁前执行一次cleanup函数

性能调优

性能方面入手无非就是两个方面：时间；空间

尽量最快执行；尽量用最少的空间

调优可从以下几个方面入手：

A.输入采用大文件

hadoop习惯处理大文件，讨厌小文件。所以在输入时尽量采用少量的大文件，避免使用大量的小文件。

B.压缩文件

对Map处理的结果进行压缩，即可减少空间的存储，又可以减少在网络中的数据传输

C.过滤数据

在处理Map前先对数据进行过滤，例如：一个表巨大，一个表比较小。当两个表做等值连接时，是不是应该采用小表里面的数据来对比大表里面数据。在这里有个Bloom Filte。大家可以自己去学习

D.调整作业属性

调整Map函数的个数和Reduce函数的个数，达到最佳

E.Combine函数

Conbine函数用于本地合并，这会大大减少网络I/O操作的消耗

评论

1 楼 annmi_cai 2016-03-11  

好好学习，天天向上！