Hadoop（十二）MapReduce概述

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)
        );;
    }
}