MapReduce学习笔记(4)

现在正式开始编写MapReduce程序。

1、专利数据集

　　这里我们使用专利数据集作为hadoop的输入数据。数据集可以从美国国家经济研究局获得:http://www.nber.org/patents/ 。两个数据集下载地址分别为：

http://www.nber.org/patents/acite75_99.zip http://www.nber.org/patents/apat63_99.zip。

2、MapReduce程序的基础模板

　　大多数MapReduce程序的编写都可以依赖于一个模板或其变种，当写程序是，我们将其修改为我们所希望的样子，而不是重新写一个。

　　我们通过一个简单的例子来给出一个简单的MapReduce程序模板。在cite75_99.txt文件中，保存了专利引用数据。CITING表示专利的编号，CITED表示被引用的专利的编号。下面我们给出一个例子来列出一个专利被哪些专利引用了。

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

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;

@SuppressWarnings("deprecation")
public class MyJob extends Configured implements Tool {
    public static class MapClass extends MapReduceBase implements Mapper<Text, Text, Text, Text>{//交换key和value的顺序，使被引用的专利号作为key
        @Override
        public void map(Text key, Text value, OutputCollector<Text, Text> output,
                Reporter reporter) throws IOException {
            output.collect(value, key);
        }    
    }
    public static class Reduceclass extends MapReduceBase implements Reducer<Text, Text, Text, Text>{//对被引用的专利进行reduce，输出所有引用它的专利
        @Override
        public void reduce(Text key, Iterator<Text> value,
                OutputCollector<Text, Text> output, Reporter repoter)
                throws IOException {
            String cvs="";
            while(value.hasNext()){
                if(cvs.length()>0)
                    cvs+=",";
                cvs+=value.next().toString();
            }
            output.collect(key, new Text(cvs));
        }
    }
    @Override
    public int run(String[] args) throws Exception {//完成任务的各种设置
        Configuration conf = getConf();
        JobConf job = new JobConf(conf,MyJob.class);
        
        Path in = new Path(args[0]);
        Path out = new Path(args[1]);
        FileInputFormat.setInputPaths(job, in);
        FileOutputFormat.setOutputPath(job, out);
        
        job.setJobName("MyJob");
        job.setMapperClass(MapClass.class);
        job.setReducerClass(Reduceclass.class);
        
        job.setInputFormat(KeyValueTextInputFormat.class);
        job.setOutputFormat(TextOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.set("key.value.separator.in.input.line", ",");
        try{
            JobClient.runJob(job);
        }catch(Exception e){
            e.printStackTrace();
        }
        return 0;
    }
    public static void main(String[] args) {
        int ret=0;
        try {
            ret = ToolRunner.run(new Configuration(),new MyJob(),args);
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            System.exit(ret);
        }
    }    
}

　　 运行前运行后

　　习惯上用单个类完整定义MapReduce作业，这里称为MyJob类。Hadoop要求Mapper和Reducer必须是它们自身的静态类。
　　

　　框架的核心在run()方法中，也称为driver。它实例化、配置并传递一个JobConf对象命名的作业给JobClient.runJOb()以启动MapReduce作业。JobConf对象保存作业运行所需要的全部配置参数。driver需要为每个作业定制基本参数，包括输入输出路径、Mapper类和Reducer类。作业可以重置默认的作业属性，如前面提到的InputFormat和OutputFormat等，也可以调用JobConf对象中的set()方法填充任意的配置参数。一旦传递JobConf对象到JobClient.runJob()，它就视为作业的整体规划，成为决定这个作业如何运行的蓝本。

　　Mapper类的核心是map()方法，Reducer类的核心是reduce()方法。每一个map()方法的调用分别被赋予一个类型为K1和V1的key/value对。这个key/value由mapper生成，并通过OutputCollector对象的collect()方法来输出，只需要在合适的位置调用output.collect((K2) k,(V2) v)（其中，output是OutputCollector的实例化对象）。

　　Reducer的输入类型必须与Mapper的输出类型相同，它也是使用OutputCollector来输出key/value对， output.collect((K3) k , (V3)v)。

　　最终，所有的key value的类型都必须是Writable的子类，来确保hadoop的序列化接口可以把数据在分布式系统上发送。