本文以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中。

  1. 通过SSH方式登录集群,详情请参见使用SSH连接主节点
  2. 执行以下命令,新建目录wordcount_classes
    mkdir wordcount_classes
  3. 执行以下命令,新建文件EmrWordCount.java
    1. 执行以下命令,打开文件EmrWordCount.java
      vim EmrWordCount.java
    2. 按下i键进入编辑模式。
    3. 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);
         }
       }
    4. 按下Esc键退出编辑模式,输入:wq保存并关闭文件。
  4. 编译并打包文件。
    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目录下。
  5. 创建作业。
    1. 步骤4wordcount.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/Inputoss://<yourBucketName>/data/WordCount/Output分别为输入输出路径。

    3. 在作业编辑中,单击运行
      MR作业就会在指定的集群中运行起来。

Wordcount2示例

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

  1. 在本地安装Maven和Java环境。
    本示例中Maven是3.0版本,Java是1.8版本。
  2. 执行如下命令,生成工程框架。
    例如,您的工程开发根目录是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一致)。

  3. 加入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>
  4. 编写代码。
    1. 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;
           }
       }
    2. 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);
           }
       }
  5. 在工程的目录下,执行如下命令,编译并打包文件。
    mvn clean package -DskipTests

    您可以在工程目录的target目录下看到名称为wordcountv2-1.0-SNAPSHOT.jar的JAR包。

  6. 创建作业。
    1. 步骤5wordcountv2-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文件用来处理需要忽略(不计频次)的单词。
    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作业就会在指定的集群中运行起来。