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.txt 
public static class PatentMapper 
extends Mapper<LongWritable,Text,Text,IntWritable>{ 
private PatentRecordParser parser = new PatentRecordParser(); 
private  Text key = new Text(); 
//把进入reduce的value都设置成1 
private IntWritable value = new IntWritable(1); 
//进入map端的数据，每次进入一行。 
//MapReduce都是具有一定结构的数据，有一定含义的数据。 
//进入时候map的k1(该行数据首个字符距离整个文档首个字符的距离),v1(这行数据的字符串) 
        @Override 
protected 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 PatentReducer 
extends Reducer<Text,IntWritable,Text,IntWritable>{ 
@Override 
protected 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上面的打印语句是没有办法输出的，但会记录到日志文件当中。 
        } 
} 
@Override 
public 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) 
);; 
} 
}