Hadoop学习01：基础程序模板

看hadoop in action这本书，写的不错，就是没有兼容新的API，有点遗憾。

第一个例子讲倒排索引问题，本质上是求一个图的入度问题，将一个有向图邻接矩阵转置对每行求和。简单的协同推荐也是这样的思想。

基本的MapReduce编程遵循一个模板。逻辑由聚合函数来描述：分配型、代数型、全集型。

复杂MapReduce作业的链接：A->B->C（顺序型）、(A, B)->C（依赖型）。

1. MapReduce基础程序

　　首先以倒排索引为例子，形成MapReduce的基本思维。采用专利引用数据集cite75_99.txt，数据格式如下：

　　　　　　　　　　　　　　"CITING","CITED"　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　 　　　　　　　 3858241,956203 　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　 ...

程序如下：

package org.apache.hadoop.examples;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MyJob extends Configured implements Tool{

public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {

public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {

String[] citation = value.toString().split(",");
context.write(new Text(citation[1]), new Text(citation[0]));

}
}

public static class Reduce extends Reducer<Text, Text, Text, Text> {

public void reduce(Text key, Iterable<Text> values,
Context context) throws IOException, InterruptedException {

String csv = "";
for (Text val : values) {

if (csv.length() > 0) csv += ",";
csv += val.toString();
}

context.write(key, new Text(csv));
}
}

public int run(String[] args) throws Exception {

Configuration conf = getConf();

Job job = new Job(conf, "MyJob");
job.setJarByClass(MyJob.class);

Path in = new Path(args[0]);
Path out = new Path(args[1]);
FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);

job.setMapperClass(MapClass.class);
job.setReducerClass(Reduce.class);

job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

System.exit(job.waitForCompletion(true) ? 0 : 1);

return 0;
}

public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new MyJob(), args);

System.exit(res);
}
}

首先，习惯上用单个类来完整的定义一个MapReduce作业。Hadoop要求Mapper和Reducer必须是它们自身的静态类，作为MyJob的内部类简化代码管理，内部类是独立的。在作业执行期间，这两个类复制并运行在各个节点的JVM上，而MyJob的其他部分只运行在客户机上。

框架的核心是run()方法。其中Job对象保存节点运行的配置参数。Hadoop框架提供了ToolRunner、Tool、Configured简化实现。　　　　　　　　　　　　　　　　　　　　

（1）MyJob对象通过继承Configured类保持默认的配置参数。

Configuration conf = getConf();
Job job = new Job(conf, "MyJob");
job.setJarByClass(MyJob.class);

（2）定制作业基本参数

　包括输入路径、输出路径、Mapper类、Reducer类。

FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);
job.setMapperClass(MapClass.class);
job.setReducerClass(Reduce.class);

（3）定制作业属性

注意：InputFormat、OutputFormat分别与Mapper的输入参数、Reducer的输入参数对应；而OutputKey、OutputValue与中间键值对对应。

job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
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处理的是<K1,V1>这样的键值对，然后抛出<K2,V2>键值对到"shuffle"阶段；“shuffle”阶段经过洗牌把相同K2对应的键值对归到同一个桶里，形成<K2,Iterable<V2>>，输入Reduce阶段；Reduce阶段处理归并的键值对，抛出<K3,V3>写入Output路径。

思考：如何迭代的进行大数据计算呢？只要将中间结果存放在input路径中，在main函数中不停的调用即可。

思考：hadoop编程的核心是什么呢？从架构上是如何可靠的分布式进行分布式计算，从算法上是如何发挥并行算法的优势。