import org.apache.hadoop.io.Text;public class PatentRecordParser {//1,2//1,3//2,3//表示数据中的第一列private String patentId;//表示数据中的第二列private String refPatentId;//表示解析的当前行的数据是否有效private boolean valid;public void parse(String line){String[]  strs = line.split(",");if (strs.length==2){patentId = strs[0].trim();refPatentId = strs[1].trim();if (patentId.length()>0&&refPatentId.length()>0){valid = true;}}}public void parse(Text line){parse(line.toString());}public String getPatentId() {return patentId;}public void setPatentId(String patentId) {this.patentId = patentId;}public String getRefPatentId() {return refPatentId;}public void setRefPatentId(String refPatentId) {this.refPatentId = refPatentId;}public boolean isValid() {return valid;}public void setValid(boolean valid) {this.valid = valid;}
}

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;import java.io.IOException;public class PatentReference_0011 extends Configured implements Tool {//-Dinput=/data/patent/cite75_99.txtpublic static class PatentMapperextends Mapper<LongWritable,Text,Text,IntWritable>{private PatentRecordParser parser = new PatentRecordParser();private  Text key = new Text();//把进入reduce的value都设置成1private IntWritable value = new IntWritable(1);//进入map端的数据，每次进入一行。//MapReduce都是具有一定结构的数据，有一定含义的数据。//进入时候map的k1(该行数据首个字符距离整个文档首个字符的距离),v1(这行数据的字符串)@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {parser.parse(value);if (parser.isValid()){this.key.set(parser.getRefPatentId());context.write(this.key,this.value);}}}public static class PatentReducerextends Reducer<Text,IntWritable,Text,IntWritable>{@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int count = 0;for (IntWritable iw:values){count+=iw.get();}context.write(key,new IntWritable(count));//注意：在map或reduce上面的打印语句是没有办法输出的，但会记录到日志文件当中。}}@Overridepublic int run(String[] args) throws Exception {//构建作业所处理的数据的输入输出路径Configuration conf = getConf();Path input = new Path(conf.get("input"));Path output = new Path(conf.get("output"));//构建作业配置Job job = Job.getInstance(conf,this.getClass().getSimpleName()+"Lance");//如果不指定取的名字就是当前类的类全名//设置该作业所要执行的类job.setJarByClass(this.getClass());//设置自定义的Mapper类以及Map端数据输出时的类型job.setMapperClass(PatentMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//设置自定义的Reducer类以及输出时的类型job.setReducerClass(PatentReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//设置读取最原始数据的格式信息以及//数据输出到HDFS集群中的格式信息job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);//设置数据读入和写出的路径到相关的Format类中TextInputFormat.addInputPath(job,input);TextOutputFormat.setOutputPath(job,output);//提交作业return job.waitForCompletion(true)?0:1;}public static void main(String[] args) throws Exception {System.exit(ToolRunner.run(new PatentReference_0011(),args));;}
}