MaxCompute用户指南：MapReduce：概要：开源兼容MapReduce

MaxCompute（原 ODPS）有一套原生的 MapReduce 编程模型和接口，简单来说，这套接口的输入输出都是MaxCompute 中的 Table，处理的数据是以 Record 为组织形式的，它可以很好地描述 Table 中的数据处理过程。但是与社区的Hadoop 相比，编程接口差异较大。Hadoop 用户如果要将原来的 Hadoop MR 作业迁移到 MaxCompute 的 MR中执行，需要重写 MR 的代码，使用 MaxCompute 的接口进行编译和调试，运行正常后再打成一个 Jar 包才能放到 MaxCompute的平台来运行。这个过程十分繁琐，需要耗费很多的开发和测试人力。如果能够完全不改或者少量地修改原来的 Hadoop MapReduce代码，便可在 MaxCompute 平台上运行，将会比较理想。
现在 MaxCompute 平台提供了一个 Hadoop MapReduce 到 MaxCompute MR的适配工具，已经在一定程度上实现了 Hadoop MapReduce作业的二进制级别的兼容，即您可以在不改代码的情况下通过指定一些配置，便可将原来在 Hadoop 上运行的 MapReduce Jar包拿过来直接运行在 MaxCompute 上。您可通过此处下载开发插件。目前该插件处于测试阶段，暂时还不能支持您自定义 comparator 和自定义 key 类型。
下面将以 WordCount 程序为例，为您介绍 HadoopMR 插件的基本使用方式。

注意：

更多开源兼容性的介绍请参见开源版本 SDK 兼容性。

更多 Hadoop MapReduce SDK 的介绍请参见 MapReduce 官方文档。

下载 HadoopMR 插件

请单击此处，下载插件，包名为 hadoop2openmr-1.0.jar。

注意：
这个 Jar 包中已经包含 hadoop-2.7.2 版本的相关依赖，在作业的 Jar 包中请不要携带 Hadoop 的依赖，避免版本冲突。

准备 Jar 包

编译导出 WordCount 的 Jar 包：wordcount_test.jar，WordCount 程序的源码如下：

package com.aliyun.odps.mapred.example.hadoop;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
import java.util.StringTokenizer;
public class WordCount {
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}
public static class IntSumReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values,
Context context
) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

准备测试数据

创建输入表和输出表。create table if not exists wc_in(line string);
create table if not exists wc_out(key string, cnt bigint);

通过 Tunnel 命令将数据导入输入表中。
需要导入文本文件 data.txt 的数据内容如下：

hello maxcompute
hello mapreduce

您可通过 MaxCompute 客户端的 Tunnel 命令将 data.txt 的数据导入 wc_in 中，如下所示：

tunnel upload data.txt wc_in;

准备好表与 HDFS 文件路径的映射关系配置

配置文件命名为：wordcount-table-res.conf

{
"file:/foo": {
"resolver": {
"resolver": "com.aliyun.odps.mapred.hadoop2openmr.resolver.TextFileResolver",
"properties": {
"text.resolver.columns.combine.enable": "true",
"text.resolver.seperator": "\t"
}
},
"tableInfos": [
{
"tblName": "wc_in",
"partSpec": {},
"label": "__default__"
}
],
"matchMode": "exact"
},
"file:/bar": {
"resolver": {
"resolver": "com.aliyun.odps.mapred.hadoop2openmr.resolver.BinaryFileResolver",
"properties": {
"binary.resolver.input.key.class" : "org.apache.hadoop.io.Text",
"binary.resolver.input.value.class" : "org.apache.hadoop.io.LongWritable"
}
},
"tableInfos": [
{
"tblName": "wc_out",
"partSpec": {},
"label": "__default__"
}
],
"matchMode": "fuzzy"
}
}

配置项说明：
整个配置是一个 Json 文件，描述 HDFS 上的文件与 MaxCompute 上的表之间的映射关系，一般要配置输入和输出两部分，一个 HDFS 路径对应一个 resolver 配置，tableInfos 配置以及 matchMode 配置。

resolver：用于配置如何对待文件中的数据，目前有com.aliyun.odps.mapred.hadoop2openmr.resolver.TextFileResolver 和com.aliyun.odps.mapred.hadoop2openmr.resolver.BinaryFileResolver 两个内置的resolver 可以选用。除了指定好 resolver 的名字，还需要为相应的 resolver 配置一些 properties指导它正确的进行数据解析。
TextFileResolver：对于纯文本的数据，输入输出都会当成纯文本对待。当作为输入 resolver 配置时，需要配置的properties 有：text.resolver.columns.combine.enable 和text.resolver.seperator，当text.resolver.columns.combine.enable 配置为 true时，会把输入表的所有列按找 text.resolver.seperator指定的分隔符组合成一个字符串作为输入。否则，会把输入表的前两列分别作为 key，value。
BinaryFileResolver：可以处理二进制的数据，自动将数据转换为 MaxCompute可以支持的数据类型，如：Bigint，Bool，Double 等。当作为输出 resolver 配置时，需要配置的 properties有：binary.resolver.input.key.class 和binary.resolver.input.value.class，分别代表中间结果的 key 和 value 类型。

tableInfos：您配置 HDFS 对应的 MaxCompute 的表，目前只支持配置表的名字 tblName，而 partSpec 和 label 请保持和示例一致。
matchMode：路径的匹配模式，可选项为 exact 和 fuzzy，分别代表精确匹配和模糊匹配，如果设置为 fuzzy，则可以通过正则来匹配 HDFS 的输入路径。

作业提交

使用 MaxCompute 命令行工具 odpscmd 提交作业。MaxCompute 命令行工具的安装和配置方法请参见客户端用户手册。在 odpscmd 下运行如下命令：

jar -DODPS_HADOOPMR_TABLE_RES_CONF=./wordcount-table-res.conf -classpath hadoop2openmr-1.0.jar,wordcount_test.jar com.aliyun.odps.mapred.example.hadoop.WordCount /foo /bar;

这里假设已经将 hadoop2openmr-1.0.jar、wordcount_test.jar 和 wordcount-table-res.conf 放置到 odpscmd 的当前目录，否则在指定配置和 -classpath 的路径时需要做相应的修改。
运行过程如下图所示：