Chapter04 编写基本的MapReduce程序(一) 专利数据集实战一

获取专利数据集的网站：http://www.nber.org/patents/

需要下载的数据集：pat63_99.txt和Cite75_99.txt

专利引用数据的格式如下，采用逗号分隔

专利描述数据集，各个字段的意义：

大体的程序如下所示：

package cn.edu.hust.job;

import java.io.IOException;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MyJob extends Configured implements Tool {

	public static class MapClass extends MapReduceBase implements Mapper<Text,Text,Text,Text>
	{

		public void map(Text key, Text value, OutputCollector<Text, Text> collect,
				Reporter report) throws IOException {
			collect.collect(value, key);
		}
		
	}
	
	
	public static class ReducerClass extends  MapReduceBase implements Reducer<Text, Text, Text, Text>
	{

		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> collect, Reporter report)
				throws IOException {
			String csv="";
			while(values.hasNext())
			{
				if (csv.length()>0) {
					csv+=",";
				}
				csv=values.next().toString();
			}
			collect.collect(key, new Text(csv));
		}
		
	}
	
	
	//核心
	public int run(String[] arg0) throws Exception {
		//创建一个作业
		Configuration conf=getConf();
		JobConf jobConf=new JobConf(conf,MyJob.class);
		//设置输入输出路径
		Path in=new Path(arg0[0]);
		Path out=new Path(arg0[1]);
		FileInputFormat.setInputPaths(jobConf,in);
		FileOutputFormat.setOutputPath(jobConf, out);
		//设置mapper、reducer对象
		jobConf.setJobName("MyJob");
		jobConf.setMapperClass(MapClass.class);
		jobConf.setReducerClass(ReducerClass.class);
		
		//指定K1、V1的数据格式
		jobConf.setInputFormat(KeyValueTextInputFormat.class);
		jobConf.setOutputFormat(TextOutputFormat.class);
		//指定K2、V2的数据格式
		jobConf.setOutputKeyClass(Text.class);
		jobConf.setOutputValueClass(Text.class);
		jobConf.set("key.value.separator.in.input.line", ",");
		
		//启动MapReduce作业
		JobClient.runJob(jobConf);
		
		return 0;
	}
	
	public static void main(String[] args) throws Exception {
		int res=ToolRunner.run(new Configuration(), new MyJob(), args);
		System.exit(res);
	}

}

运行的过程：

得到的部分结果：