MapReduce 初级编程实践

（一）编程实现文件合并和去重操作**

对于两个输入文件，即文件 A 和文件 B，请编写 MapReduce 程序，对两个文件进行合并， 并剔除其中重复的内容，得到一个新的输出文件 C。下面是输入文件和输出文件的一个样例供参考。

输入文件 A 的样例如下：

20170101 x 
20170102 y
20170103 x
20170104 y
20170105 z
20170106 x

输入文件 B的样例如下：

20170101 y
20170102 y
20170103 x
20170104 z
20170105 y

根据输入文件 A 和 B 合并得到的输出文件 C 的样例如下：

20170101 x
20170101 y
20170102 y
20170103 x
20170104 y
20170104 z
20170105 y
20170105 z
20170106 x

启动hadoop：

cd /usr/local/hadoop
./sbin/start-dfs.sh

新建input文件夹，向hdfs上传文件，将家目录下的A.txt和B.txt上传到hdfs的/user/hadoop/input下

./bin/hdfs dfs -mkdir input
./bin/hdfs dfs -ls
./bin/hdfs dfs -put ~/A.txt input
./bin/hdfs dfs -put ~/B.txt input
./bin/hdfs dfs -ls input

启动eclipse，编程实现文件合并和去重操作：

package mapReduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MergeHeavy {
   
	
	public static class Map extends Mapper<Object, Text, Text, Text>{
   
		private static Text text = new Text();
		public void map(Object key, Text value, Context context) throws IOException,InterruptedException{
   
			text = value;
			context.write(text, new Text(""));
		}
	}
	
	public static class Reduce extends Reducer<Text, Text, Text, Text>{
   
		public void reduce(Text key, Iterable<Text> values, Context context ) throws IOException,InterruptedException{
   
			context.write(key, new Text(""));
		}
	}
	
	public static void main(String[] args) throws Exception{
   
		
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		conf.set("fs.default.name","hdfs://localhost:9000");
		String[] otherArgs = new String[]{
   "input","output"}; 
		if (otherArgs.length != 2) {
   
			System.err.println("Usage: wordcount <in><out>");
			System.exit(2);
			}
		Job job = Job.getInstance(conf,"Merge and duplicate removal");//设置环境参数
		job.setJarByClass(MergeHeavy.class);
		job.setMapperClass(Map.class);
		job.setCombinerClass(Reduce.class);
		job.setReducerClass(Reduce.class);
		job.setOutputKeyClass(Text.class);//设置输出类型
		job.setOutputValue