MapReduce开发手册 - E-MapReduce

本文以EMR-3.27.0集群为例，通过以下示例为您介绍如何在E-MapReduce集群中开发MR作业。

在MapReduce中使用OSS

在MapReduce中读写OSS，需要配置如下参数。

conf.set("fs.oss.accessKeyId", "${accessKeyId}");
conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
conf.set("fs.oss.endpoint","${endpoint}");

参数说明：

${accessKeyId}：阿里云账号的AccessKey ID。
${accessKeySecret}：阿里云账号的AccessKey Secret。
${endpoint}：OSS对外服务的访问域名。
由您集群所在的地域决定，对应的OSS也需要是在集群对应的地域，详情请参见访问域名和数据中心。

WordCount示例

本示例为您介绍MapReduce如何从Master节点的OSS中读取文本，然后统计其中单词的数量并将数据写回到OSS中。

通过SSH方式登录集群，详情请参见使用SSH连接主节点。
执行以下命令，新建目录wordcount_classes。
```
mkdir wordcount_classes
```

执行以下命令，新建文件EmrWordCount.java。

执行以下命令，打开文件EmrWordCount.java。
```
vim EmrWordCount.java
```
按下i键进入编辑模式。

在EmrWordCount.java文件中添加以下信息。

package org.apache.hadoop.examples;
 import java.io.IOException;
 import java.util.StringTokenizer;
 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 org.apache.hadoop.util.GenericOptionsParser;
 public class EmrWordCount {
  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();
     String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
     if (otherArgs.length < 2) {
       System.err.println("Usage: wordcount <in> [<in>...] <out>");
       System.exit(2);
     }
     conf.set("fs.oss.accessKeyId", "${accessKeyId}");
     conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
     conf.set("fs.oss.endpoint","${endpoint}");
     Job job = Job.getInstance(conf, "word count");
     job.setJarByClass(EmrWordCount.class);
     job.setMapperClass(TokenizerMapper.class);
     job.setCombinerClass(IntSumReducer.class);
     job.setReducerClass(IntSumReducer.class);
     job.setOutputKeyClass(Text.class);
     job.setOutputValueClass(IntWritable.class);
     for (int i = 0; i < otherArgs.length - 1; ++i) {
       FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
     }
     FileOutputFormat.setOutputPath(job,
       new Path(otherArgs[otherArgs.length - 1]));
     System.exit(job.waitForCompletion(true) ? 0 : 1);
   }
 }

按下Esc键退出编辑模式，输入:wq保存并关闭文件。

编译并打包文件。
1. 执行以下命令，编译程序。
```
javac -classpath <HADOOP_HOME>/share/hadoop/common/hadoop-common-X.X.X.jar:<HADOOP_HOME>/share/hadoop/mapreduce/hadoop-mapreduce-client-core-X.X.X.jar:<HADOOP_HOME>/share/hadoop/common/lib/commons-cli-1.2.jar -d wordcount_classes EmrWordCount.java
```
  - HADOOP_HOME：Hadoop的安装目录，通常Hadoop的安装目录为/usr/lib/hadoop-current。
    您也可以通过env |grep hadoop命令获取安装目录。
  - X.X.X：JAR包的具体版本号，需要根据实际集群中Hadoop的版本来修改。
    hadoop-common-X.X.X.jar，您可以在<HADOOP_HOME>/share/hadoop/common/目录下查看。hadoop-mapreduce-client-core-X.X.X.jar，您可以在<HADOOP_HOME>/share/hadoop/mapreduce/目录下查看。
2. 执行以下命令，打包JAR文件。
```
jar cvf wordcount.jar -C wordcount_classes .
```
  说明本示例中，打包后的wordcount.jar文件默认保存在/root目录下。
创建作业。
1. 将步骤4的wordcount.jar上传到OSS，详情请参见上传文件。
  例如，本示例中JAR文件在OSS上的路径为oss://<yourBucketName>/jars/wordcount.jar。
2. 在E-MapReduce中新建MR作业，详情请参见Hadoop MapReduce作业配置。
  作业内容如下所示：
```
ossref://<yourBucketName>/jars/wordcount.jar org.apache.hadoop.examples.EmrWordCount oss://<yourBucketName>/data/WordCount/Input oss://<yourBucketName>/data/WordCount/Output
```
  代码中的<yourBucketName>需要替换为您实际的OSS Bucket，oss://<yourBucketName>/data/WordCount/Input和oss://<yourBucketName>/data/WordCount/Output分别为输入输出路径。
3. 在作业编辑中，单击运行。
  MR作业就会在指定的集群中运行起来。

Wordcount2示例

当您的工程规模比较大时，您可以使用类似Maven的项目管理工具来进行管理。本示例为您介绍如何通过Maven工程来管理MR作业。

在本地安装Maven和Java环境。
本示例中Maven是3.0版本，Java是1.8版本。
执行如下命令，生成工程框架。
例如，您的工程开发根目录是D:/workspace。
```
mvn archetype:generate -DgroupId=com.aliyun.emr.hadoop.examples -DartifactId=wordcountv2 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false
```
通过以上命令会自动生成一个空的Sample工程位于D:/workspace/wordcountv2（和您指定的artifactId一致），里面包含一个简单的pom.xml和App类（类的包路径和您指定的groupId一致）。

加入Hadoop依赖。

使用IDE打开Sample工程，编辑pom.xml文件，当Hadoop是2.8.5版本时，需要添加如下内容。

<dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-mapreduce-client-common</artifactId>
         <version>2.8.5</version>
</dependency>
<dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-common</artifactId>
         <version>2.8.5</version>
</dependency>

编写代码。

在com.aliyun.emr.hadoop.examples中和App类平行的位置添加新类EMapReduceOSSUtil。

package com.aliyun.emr.hadoop.examples;
 import org.apache.hadoop.conf.Configuration;
 public class EMapReduceOSSUtil {
     private static String SCHEMA = "oss://";
     private static String EPSEP = ".";
     private static String HTTP_HEADER = "http://";
     /**
      * complete OSS uri
      * convert uri like: oss://bucket/path to oss://bucket.endpoint/path
      * ossref do not need this
      *
      * @param oriUri original OSS uri
      */
     public static String buildOSSCompleteUri(String oriUri, String endpoint) {
         if (endpoint == null) {
             System.err.println("miss endpoint");
             return oriUri;
         }
         int index = oriUri.indexOf(SCHEMA);
         if (index == -1 || index != 0) {
             return oriUri;
         }
         int bucketIndex = index + SCHEMA.length();
         int pathIndex = oriUri.indexOf("/", bucketIndex);
         String bucket = null;
         if (pathIndex == -1) {
             bucket = oriUri.substring(bucketIndex);
         } else {
             bucket = oriUri.substring(bucketIndex, pathIndex);
         }
         StringBuilder retUri = new StringBuilder();
         retUri.append(SCHEMA)
                 .append(bucket)
                 .append(EPSEP)
                 .append(stripHttp(endpoint));
         if (pathIndex > 0) {
             retUri.append(oriUri.substring(pathIndex));
         }
         return retUri.toString();
     }
     public static String buildOSSCompleteUri(String oriUri, Configuration conf) {
         return buildOSSCompleteUri(oriUri, conf.get("fs.oss.endpoint"));
     }
     private static String stripHttp(String endpoint) {
         if (endpoint.startsWith(HTTP_HEADER)) {
             return endpoint.substring(HTTP_HEADER.length());
         }
         return endpoint;
     }
 }

在com.aliyun.emr.hadoop.examples中和App类平行的位置添加新类WordCount2.java。

package com.aliyun.emr.hadoop.examples;
 import java.io.BufferedReader;
 import java.io.FileReader;
 import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.StringTokenizer;
 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 org.apache.hadoop.mapreduce.Counter;
 import org.apache.hadoop.util.GenericOptionsParser;
 import org.apache.hadoop.util.StringUtils;
 public class WordCount2 {
     public static class TokenizerMapper
             extends Mapper<Object, Text, Text, IntWritable>{
         static enum CountersEnum { INPUT_WORDS }
         private final static IntWritable one = new IntWritable(1);
         private Text word = new Text();
         private boolean caseSensitive;
         private Set<String> patternsToSkip = new HashSet<String>();
         private Configuration conf;
         private BufferedReader fis;
         @Override
         public void setup(Context context) throws IOException,
                 InterruptedException {
             conf = context.getConfiguration();
             caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
             if (conf.getBoolean("wordcount.skip.patterns", true)) {
                 URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
                 for (URI patternsURI : patternsURIs) {
                     Path patternsPath = new Path(patternsURI.getPath());
                     String patternsFileName = patternsPath.getName().toString();
                     parseSkipFile(patternsFileName);
                 }
             }
         }
         private void parseSkipFile(String fileName) {
             try {
                 fis = new BufferedReader(new FileReader(fileName));
                 String pattern = null;
                 while ((pattern = fis.readLine()) != null) {
                     patternsToSkip.add(pattern);
                 }
             } catch (IOException ioe) {
                 System.err.println("Caught exception while parsing the cached file '"
                         + StringUtils.stringifyException(ioe));
             }
         }
         @Override
         public void map(Object key, Text value, Context context
         ) throws IOException, InterruptedException {
             String line = (caseSensitive) ?
                     value.toString() : value.toString().toLowerCase();
             for (String pattern : patternsToSkip) {
                 line = line.replaceAll(pattern, "");
             }
             StringTokenizer itr = new StringTokenizer(line);
             while (itr.hasMoreTokens()) {
                 word.set(itr.nextToken());
                 context.write(word, one);
                 Counter counter = context.getCounter(CountersEnum.class.getName(),
                         CountersEnum.INPUT_WORDS.toString());
                 counter.increment(1);
             }
         }
     }
     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();
         conf.set("fs.oss.accessKeyId", "${accessKeyId}");
         conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
         conf.set("fs.oss.endpoint","${endpoint}");
         GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
         String[] remainingArgs = optionParser.getRemainingArgs();
         if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
             System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
             System.exit(2);
         }
         Job job = Job.getInstance(conf, "word count");
         job.setJarByClass(WordCount2.class);
         job.setMapperClass(TokenizerMapper.class);
         job.setCombinerClass(IntSumReducer.class);
         job.setReducerClass(IntSumReducer.class);
         job.setOutputKeyClass(Text.class);
         job.setOutputValueClass(IntWritable.class);
         List<String> otherArgs = new ArrayList<String>();
         for (int i=0; i < remainingArgs.length; ++i) {
             if ("-skip".equals(remainingArgs[i])) {
                 job.addCacheFile(new Path(EMapReduceOSSUtil.buildOSSCompleteUri(remainingArgs[++i], conf)).toUri());
                 job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
             } else {
                 otherArgs.add(remainingArgs[i]);
             }
         }
         FileInputFormat.addInputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(0), conf)));
         FileOutputFormat.setOutputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(1), conf)));
         System.exit(job.waitForCompletion(true) ? 0 : 1);
     }
 }

在工程的目录下，执行如下命令，编译并打包文件。
```
mvn clean package -DskipTests
```
您可以在工程目录的target目录下看到名称为wordcountv2-1.0-SNAPSHOT.jar的JAR包。
创建作业。
1. 将步骤5的wordcountv2-1.0-SNAPSHOT.jar上传到OSS，详情请参见上传文件。
  例如，本示例中JAR文件在OSS上的路径为oss://<yourBucketName>/jars/wordcountv2-1.0-SNAPSHOT.jar。
2. 下载并上传以下文件至您OSS的对应目录。
  - The_Sorrows_of_Young_Werther.txt
  - patterns.txt
  说明 The_Sorrows_of_Young_Werther.txt为待统计单词的文本文件，patterns.txt文件用来处理需要忽略（不计频次）的单词。
3. 在E-MapReduce中新建MR作业，详情请参见Hadoop MapReduce作业配置。
  作业内容如下所示：
```
ossref://<yourBucketName>/jars/wordcountv2-1.0-SNAPSHOT.jar com.aliyun.emr.hadoop.examples.WordCount2 -D wordcount.case.sensitive=true oss://<yourBucketName>/jars/The_Sorrows_of_Young_Werther.txt oss://<yourBucketName>/jars/output -skip oss://<yourBucketName>/jars/patterns.txt
```
  代码中的<yourBucketName>需要替换为您实际的OSS Bucket，输出路径为oss://<yourBucketName>/jars/output。
4. 在作业编辑中，单击运行。
  MR作业就会在指定的集群中运行起来。