Hadoop学习01：基础程序模板

darkpoon

　　看hadoop in action这本书，写的不错，就是没有兼容新的API，有点遗憾。
　　第一个例子讲倒排索引问题，本质上是求一个图的入度问题，将一个有向图邻接矩阵转置对每行求和。简单的协同推荐也是这样的思想。
　　基本的MapReduce编程遵循一个模板。逻辑由聚合函数来描述：分配型、代数型、全集型。
　　复杂MapReduce作业的链接：A->B->C（顺序型）、(A, B)->C（依赖型）。
　　1. MapReduce基础程序
　　首先以倒排索引为例子，形成MapReduce的基本思维。采用专利引用数据集cite75_99.txt，数据格式如下：
　　"CITING","CITED"　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　   　　　　　　　  3858241,956203 　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　     ...
　　程序如下：



1 package org.apache.hadoop.examples;
2
3 import java.io.IOException;
4
5 import org.apache.hadoop.conf.Configuration;
6 import org.apache.hadoop.conf.Configured;
7 import org.apache.hadoop.fs.Path;
8 import org.apache.hadoop.io.LongWritable;
9 import org.apache.hadoop.io.Text;
10 import org.apache.hadoop.mapreduce.lib.input.*;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
13 import org.apache.hadoop.mapreduce.Job;
14 import org.apache.hadoop.mapreduce.Mapper;
15 import org.apache.hadoop.mapreduce.Reducer;
16 import org.apache.hadoop.util.Tool;
17 import org.apache.hadoop.util.ToolRunner;
18
19 public class MyJob extends Configured implements Tool{
20
21     public static class MapClass extends Mapper {
22         
23         public void map(LongWritable key, Text value, Context context)
24                         throws IOException, InterruptedException {
25            
26             String[] citation = value.toString().split(",");
27             context.write(new Text(citation[1]), new Text(citation[0]));
28            
29         }
30     }
31     
32     public static class Reduce extends Reducer {
33         
34         public void reduce(Text key, Iterable values,
35                 Context context) throws IOException, InterruptedException {
36            
37             String csv = "";
38             for (Text val : values) {
39                 
40                 if (csv.length() > 0) csv += ",";
41                 csv += val.toString();
42             }
43            
44             context.write(key, new Text(csv));
45         }
46     }
47     
48     public int run(String[] args) throws Exception {
49         
50         Configuration conf = getConf();
51         
52         Job job = new Job(conf, "MyJob");
53         job.setJarByClass(MyJob.class);
54         
55         Path in = new Path(args[0]);
56         Path out = new Path(args[1]);
57         FileInputFormat.setInputPaths(job, in);
58         FileOutputFormat.setOutputPath(job, out);
59         
60         job.setMapperClass(MapClass.class);
61         job.setReducerClass(Reduce.class);
62         
63         job.setInputFormatClass(TextInputFormat.class);
64         job.setOutputFormatClass(TextOutputFormat.class);
65         job.setOutputKeyClass(Text.class);
66         job.setOutputValueClass(Text.class);
67         
68         System.exit(job.waitForCompletion(true) ? 0 : 1);
69         
70         return 0;
71     }
72     
73     public static void main(String[] args) throws Exception {
74         int res = ToolRunner.run(new Configuration(), new MyJob(), args);
75         
76         System.exit(res);
77     }
78 }
　　首先，习惯上用单个类来完整的定义一个MapReduce作业。Hadoop要求Mapper和Reducer必须是它们自身的静态类，作为MyJob的内部类简化代码管理，内部类是独立的。在作业执行期间，这两个类复制并运行在各个节点的JVM上，而MyJob的其他部分只运行在客户机上。
　　框架的核心是run()方法。其中Job对象保存节点运行的配置参数。Hadoop框架提供了ToolRunner、Tool、Configured简化实现。　　　　　　　　　　　　　　　　　　　　
　　（1）MyJob对象通过继承Configured类保持默认的配置参数。



1 Configuration conf = getConf();
2 Job job = new Job(conf, "MyJob");
3 job.setJarByClass(MyJob.class);
　　（2）定制作业基本参数
　　包括输入路径、输出路径、Mapper类、Reducer类。



1 FileInputFormat.setInputPaths(job, in);
2 FileOutputFormat.setOutputPath(job, out);
3 job.setMapperClass(MapClass.class);
4 job.setReducerClass(Reduce.class);
　　（3）定制作业属性
　　注意：InputFormat、OutputFormat分别与Mapper的输入参数、Reducer的输入参数对应；而OutputKey、OutputValue与中间键值对对应。



1 job.setInputFormatClass(TextInputFormat.class);
2 job.setOutputFormatClass(TextOutputFormat.class);
3 job.setOutputKeyClass(Text.class);
4 job.setOutputValueClass(Text.class);
　　（4）启动作业
　　调用JobConf对象的waitForCompletion()方法启动作业。迭代过程在main方法中确定。
　　最后讲讲如何编译执行一个作业。第一步是编译并打包。
　　编译：javac -classpath [包路径] -d [.class生成路径] [源路径]
　　打包：jar -cvf [.jar生成路径] -C [.class路径] . (**最后一个点不能忘记)
　　执行：hadoop jar [.jar路径] [包内路径] [输入] [输出]
　　2.Mapper和Reducer

　　第一张图说明作业的运行，守护进程JobTracker运行在集群的主节点上，是应用程序和Hadoop之间的纽带；TaskTracker管理从节点上的Map和Reduce，并与JobTracker通信。

　　第二张图显示集群的拓扑结构。NameNode和JobTracker是主端，DataNode和TaskTracker是从端。

　　第三张图是普通MapReduce数据流，首先数据被分配到不同的节点上经过Map阶段，Map处理的是这样的键值对，然后抛出键值对到"shuffle"阶段；“shuffle”阶段经过洗牌把相同K2对应的键值对归到同一个桶里，形成，输入Reduce阶段；Reduce阶段处理归并的键值对，抛出写入Output路径。
　　思考：如何迭代的进行大数据计算呢？只要将中间结果存放在input路径中，在main函数中不停的调用即可。
　　思考：hadoop编程的核心是什么呢？从架构上是如何可靠的分布式进行分布式计算，从算法上是如何发挥并行算法的优势。