169      人物

開源兼容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程序為例，介紹一下這個插件的基本使用方式。

提示：

• 開源版本SDK下載地址，請點擊這裏。
• 更多開源兼容性的整理請參考開源版本SDK兼容性。
• 更多Hadoop MapReduce SDK介紹請參考MapReduce官方文檔

下載HadoopMR的插件

通過這裏下載插件，包名為hadoop2openmr-1.0.jar。

注意，這個jar裏麵已經包含hadoop-2.7.2版本的相關依賴，在作業的jar包中請不要攜帶hadoop的依賴，避免版本衝突。

準備好HadoopMR的程序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);
    }
}

hello maxcompute
hello mapreduce

tunnel upload  data.txt wc_in;

{
  "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"
  }
}

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;