Hadoop进级教程之-MapReduce代码实战

MapReduce作为一种分布式计算模型，它主要用于解决海量数据的计算问题。使用MapReduce操作海量数据时，每个MapReduce程序被初始化为一个工作任务，每个工作任务可以分为Map和Reduce两个阶段。

MapReduce计算模型主要由三个阶段组成：Map阶段、Shuffle阶段、Reduce阶段；

map阶段

（1）、将文件拆分成splits

MapReduce将文件拆分成splits，每个文件为一个split，并将文件按行分割形成对，这一步由MapReduce框架自动完成，其中偏移量(即key值)包括了回车所占的字符数(Windows和Linux环境会不同)

将分割好的对交给用户定义的Map方法进行处理，生成新的对

代码如下：

/** * map阶段 KEYIN 输入数据的key VALUEIN 输入数据的value KEYOUT 输出数据的key的类型 mastercode985,1 code,1 VALUEOUT 输出的数据的value类型 * @author www.mastercode985.top@孙俊伟 */public class WordcountMapper extends Mapper{ //定义Key和Value的存储类型及默认值 Text k = new Text(); IntWritable v = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { System.out.println(key.toString()); // 1 读取每一行数据 String line = value.toString(); // 2 切割单词 String[] words = line.split(" "); // 3 循环写出 for (String word : words) { k.set(word); context.write(k, v); } }}

（2）、Map端的shuffle，执行Combine过程

得到Map方法输出的对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果

代码如下：

/** * 每个区中重复的单词合并，并统计sum总数 * @author www.mastercode985.top@孙俊伟 */public class WordcountCombiner extends Reducer{ IntWritable v = new IntWritable(); @Override protected void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException { int sum = 0; // 1 累加求和 for (IntWritable value : values) { sum += value.get(); } v.set(sum); // 2 写出 context.write(key, v); }}

reduce阶段

Reducer先对从Mapper接收的数据进行排序，再交由用户自定义的Reduce方法进行处理，得到新的对，并作为WordCount的输出结果

代码如下：

// KEYIN, VALUEIN map阶段输出的key和valuepublic class WordcountReducer extends Reducer{ IntWritable v = new IntWritable(); @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; // 1 累加求和 for (IntWritable value : values) { sum += value.get(); } v.set(sum); context.write(key, v); }}

演示代码如下：

import java.io.IOException;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.io.compress.BZip2Codec;import org.apache.hadoop.io.compress.CompressionCodec;import org.apache.hadoop.io.compress.GzipCodec;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;//import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordcountDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { //args = new String[] { "D:/hdfstest/input", "D:/hdfstest/output/output001" }; args = new String[] { "/mastercode985/input", "/mastercode985/output/output001" };

Configuration conf = new Configuration(); // 开启map端输出压缩 conf.setBoolean("mapreduce.map.output.compress", true); // 设置map端输出压缩方式 conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class); // 1 获取Job对象 Job job = Job.getInstance(conf); // 2 设置jar存储位置 job.setJarByClass(WordcountDriver.class); // 3 关联Map和Reduce类 job.setMapperClass(WordcountMapper.class); job.setReducerClass(WordcountReducer.class); // 4 设置Mapper阶段输出数据的key和value类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5 设置最终数据输出的key和value类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class);

// 如果不设置InputFormat，它默认用的是TextInputFormat.class // job.setInputFormatClass(CombineTextInputFormat.class); // 虚拟存储切片最大值设置4m // CombineTextInputFormat.setMaxInputSplitSize(job, 4194304); // 虚拟存储切片最大值设置20m // CombineTextInputFormat.setMaxInputSplitSize(job, 20971520); // job.setNumReduceTasks(2); // job.setCombinerClass(WordcountCombiner.class); // job.setCombinerClass(WordcountReducer.class); // 设置reduce端输出压缩开启 FileOutputFormat.setCompressOutput(job, true); // 设置压缩的方式 //FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); // 6 设置输入路径和输出路径 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 提交job // job.submit(); boolean result = job.waitForCompletion(true);

System.exit(result ? 0 : 1); }}

测试阶段

(1)、打包

（2）、将打包文件复制到linux系统中

（3）、按照文件中指定的目录，创建文件路径

（4）、执行计算程序