欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

MapReduce 开发手册

程序员文章站 2024-03-24 15:35:16
...

MapReduce 开发手册


在 MapReduce 中使用 OSS

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

    

  1.     conf.set("fs.oss.accessKeyId", "${accessKeyId}");
    2. 

  2.     conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
    3. 

  3.     conf.set("fs.oss.endpoint","${endpoint}");
    4.

参数说明：

${accessKeyId}： 您账号的 AccessKeyId。

${accessKeySecret}： 该 AccessKeyId 对应的**。

${endpoint}： 访问 OSS 使用的网络，由您集群所在的 region 决定，当然对应的 OSS 也需要是在集群对应的 region。

具体的值请参考 OSS Endpoint

Word Count

以下示例介绍了如何从 OSS 中读取文本，然后统计其中单词的数量。其操作步骤如下：

  1. 程序编写。以 JAVA 代码为例，将 Hadoop 官网 WordCount 例子做如下修改。对该实例的修改只是在代码中添加了 Access Key ID 和 Access Key Secret 的配置，以便作业有权限访问 OSS 文件。

      

    1.  package org.apache.hadoop.examples;
      2. 

      3. 

    3.  import java.io.IOException;
      4. 

    4.  import java.util.StringTokenizer;
      5. 

      6. 

    6.  import org.apache.hadoop.conf.Configuration;
      7. 

    7.  import org.apache.hadoop.fs.Path;
      8. 

    8.  import org.apache.hadoop.io.IntWritable;
      9. 

    9.  import org.apache.hadoop.io.Text;
      10. 

    10.  import org.apache.hadoop.mapreduce.Job;
      11. 

    11.  import org.apache.hadoop.mapreduce.Mapper;
      12. 

    12.  import org.apache.hadoop.mapreduce.Reducer;
      13. 

    13.  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
      14. 

    14.  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
      15. 

    15.  import org.apache.hadoop.util.GenericOptionsParser;
      16. 

      17. 

    17.  public class EmrWordCount {
      18. 

      19. 

    19.   public static class TokenizerMapper
      20. 

    20.         extends Mapper<Object, Text, Text, IntWritable>{
      21. 

      22. 

    22.      private final static IntWritable one = new IntWritable(1);
      23. 

    23.      private Text word = new Text();
      24. 

      25. 

    25.      public void map(Object key, Text value, Context context
      26. 

    26.                      ) throws IOException, InterruptedException {
      27. 

    27.        StringTokenizer itr = new StringTokenizer(value.toString());
      28. 

    28.        while (itr.hasMoreTokens()) {
      29. 

    29.          word.set(itr.nextToken());
      30. 

    30.          context.write(word, one);
      31. 

    31.        }
      32. 

    32.      }
      33. 

    33.    }
      34. 

      35. 

    35.    public static class IntSumReducer
      36. 

    36.         extends Reducer<Text,IntWritable,Text,IntWritable> {
      37. 

    37.      private IntWritable result = new IntWritable();
      38. 

      39. 

    39.      public void reduce(Text key, Iterable<IntWritable> values,
      40. 

    40.                         Context context
      41. 

    41.                         ) throws IOException, InterruptedException {
      42. 

    42.        int sum = 0;
      43. 

    43.        for (IntWritable val : values) {
      44. 

    44.          sum += val.get();
      45. 

    45.        }
      46. 

    46.        result.set(sum);
      47. 

    47.        context.write(key, result);
      48. 

    48.      }
      49. 

    49.    }
      50. 

      51. 

    51.    public static void main(String[] args) throws Exception {
      52. 

    52.      Configuration conf = new Configuration();
      53. 

    53.      String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
      54. 

    54.      if (otherArgs.length < 2) {
      55. 

    55.        System.err.println("Usage: wordcount <in> [<in>...] <out>");
      56. 

    56.        System.exit(2);
      57. 

    57.      }
      58. 

      59. 

    59.      conf.set("fs.oss.accessKeyId", "${accessKeyId}");
      60. 

    60.      conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
      61. 

    61.      conf.set("fs.oss.endpoint","${endpoint}");
      62. 

      63. 

    63.      Job job = Job.getInstance(conf, "word count");
      64. 

    64.      job.setJarByClass(EmrWordCount.class);
      65. 

    65.      job.setMapperClass(TokenizerMapper.class);
      66. 

    66.      job.setCombinerClass(IntSumReducer.class);
      67. 

    67.      job.setReducerClass(IntSumReducer.class);
      68. 

    68.      job.setOutputKeyClass(Text.class);
      69. 

    69.      job.setOutputValueClass(IntWritable.class);
      70. 

    70.      for (int i = 0; i < otherArgs.length - 1; ++i) {
      71. 

    71.        FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
      72. 

    72.      }
      73. 

    73.      FileOutputFormat.setOutputPath(job,
      74. 

    74.        new Path(otherArgs[otherArgs.length - 1]));
      75. 

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

    76.    }
      77. 

    77.  }
      78.

  2. 编译程序。首先要将 jdk 和 Hadoop 环境配置好，然后进行如下操作：

      

    1.  mkdir wordcount_classes
      2. 

    2.  javac -classpath ${HADOOP_HOME}/share/hadoop/common/hadoop-common-2.6.0.jar:${HADOOP_HOME}/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar:${HADOOP_HOME}/share/hadoop/common/lib/commons-cli-1.2.jar -d wordcount_classes EmrWordCount.java
      3. 

    3.  jar cvf wordcount.jar -C wordcount_classes .
      4.

  3. 创建作业。

    • 将上一步打好的 jar 文件上传到 OSS，具体可登录 OSS 官网进行操作。假设 jar 文件在 OSS 上的路径为 oss://emr/jars/wordcount.jar, 输入输出路径分别为 oss://emr/data/WordCount/Input 和 oss://emr/data/WordCount/Output。

    • 在 E-MapReduce作业 中创建如下作业：

      MapReduce 开发手册

  4. 创建执行计划。在 E-MapReduce 执行计划中创建执行计划，将上一步创建好的作业添加到执行计划中，策略选择“立即执行”，这样 wordcount 作业就会在选定集群中运行起来了。

使用 Maven 工程来管理 MR 作业

当您的工程规模越来越大时，会变得非常复杂，不易管理。我们推荐你采用类似 Maven 这样的软件项目管理工具来进行管理。其操作步骤如下：

  1. 安装 Maven。首先确保您已经安装了 Maven。

  2. 生成工程框架。在您的工程根目录处（假设您的工程开发根目录位置是 D:/workspace）执行如下命令：

    1.  mvn archetype:generate -DgroupId=com.aliyun.emr.hadoop.examples -DartifactId=wordcountv2 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

    mvn 会自动生成一个空的 Sample 工程位于 D:/workspace/wordcountv2（和您指定的 artifactId 一致），里面包含一个简单的 pom.xml 和 App 类（类的包路径和您指定的 groupId 一致）。

  3. 加入 Hadoop 依赖。使用任意 IDE 打开这个工程，编辑 pom.xml 文件。在 dependencies 内添加如下内容：

      

    1.      <dependency>
      2. 

    2.          <groupId>org.apache.hadoop</groupId>
      3. 

    3.          <artifactId>hadoop-mapreduce-client-common</artifactId>
      4. 

    4.          <version>2.6.0</version>
      5. 

    5.      </dependency>
      6. 

    6.      <dependency>
      7. 

    7.          <groupId>org.apache.hadoop</groupId>
      8. 

    8.          <artifactId>hadoop-common</artifactId>
      9. 

    9.          <version>2.6.0</version>
      10. 

    10.      </dependency>
      11.

  4. 编写代码。在 com.aliyun.emr.hadoop.examples 包下和 App 类平行的位置添加新类 WordCount2.java。内容如下：

      

    1.  package com.aliyun.emr.hadoop.examples;
      2. 

      3. 

    3.  import java.io.BufferedReader;
      4. 

    4.  import java.io.FileReader;
      5. 

    5.  import java.io.IOException;
      6. 

    6.  import java.net.URI;
      7. 

    7.  import java.util.ArrayList;
      8. 

    8.  import java.util.HashSet;
      9. 

    9.  import java.util.List;
      10. 

    10.  import java.util.Set;
      11. 

    11.  import java.util.StringTokenizer;
      12. 

      13. 

    13.  import org.apache.hadoop.conf.Configuration;
      14. 

    14.  import org.apache.hadoop.fs.Path;
      15. 

    15.  import org.apache.hadoop.io.IntWritable;
      16. 

    16.  import org.apache.hadoop.io.Text;
      17. 

    17.  import org.apache.hadoop.mapreduce.Job;
      18. 

    18.  import org.apache.hadoop.mapreduce.Mapper;
      19. 

    19.  import org.apache.hadoop.mapreduce.Reducer;
      20. 

    20.  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
      21. 

    21.  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
      22. 

    22.  import org.apache.hadoop.mapreduce.Counter;
      23. 

    23.  import org.apache.hadoop.util.GenericOptionsParser;
      24. 

    24.  import org.apache.hadoop.util.StringUtils;
      25. 

      26. 

    26.  public class WordCount2 {
      27. 

      28. 

    28.      public static class TokenizerMapper
      29. 

    29.              extends Mapper<Object, Text, Text, IntWritable>{
      30. 

      31. 

    31.          static enum CountersEnum { INPUT_WORDS }
      32. 

      33. 

    33.          private final static IntWritable one = new IntWritable(1);
      34. 

    34.          private Text word = new Text();
      35. 

      36. 

    36.          private boolean caseSensitive;
      37. 

    37.          private Set<String> patternsToSkip = new HashSet<String>();
      38. 

      39. 

    39.          private Configuration conf;
      40. 

    40.          private BufferedReader fis;
      41. 

      42. 

    42.          @Override
      43. 

    43.          public void setup(Context context) throws IOException,
      44. 

    44.                  InterruptedException {
      45. 

    45.              conf = context.getConfiguration();
      46. 

    46.              caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
      47. 

    47.              if (conf.getBoolean("wordcount.skip.patterns", true)) {
      48. 

    48.                  URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
      49. 

    49.                  for (URI patternsURI : patternsURIs) {
      50. 

    50.                      Path patternsPath = new Path(patternsURI.getPath());
      51. 

    51.                      String patternsFileName = patternsPath.getName().toString();
      52. 

    52.                      parseSkipFile(patternsFileName);
      53. 

    53.                  }
      54. 

    54.              }
      55. 

    55.          }
      56. 

      57. 

    57.          private void parseSkipFile(String fileName) {
      58. 

    58.              try {
      59. 

    59.                  fis = new BufferedReader(new FileReader(fileName));
      60. 

    60.                  String pattern = null;
      61. 

    61.                  while ((pattern = fis.readLine()) != null) {
      62. 

    62.                      patternsToSkip.add(pattern);
      63. 

    63.                  }
      64. 

    64.              } catch (IOException ioe) {
      65. 

    65.                  System.err.println("Caught exception while parsing the cached file '"
      66. 

    66.                          + StringUtils.stringifyException(ioe));
      67. 

    67.              }
      68. 

    68.          }
      69. 

      70. 

    70.          @Override
      71. 

    71.          public void map(Object key, Text value, Context context
      72. 

    72.          ) throws IOException, InterruptedException {
      73. 

    73.              String line = (caseSensitive) ?
      74. 

    74.                      value.toString() : value.toString().toLowerCase();
      75. 

    75.              for (String pattern : patternsToSkip) {
      76. 

    76.                  line = line.replaceAll(pattern, "");
      77. 

    77.              }
      78. 

    78.              StringTokenizer itr = new StringTokenizer(line);
      79. 

    79.              while (itr.hasMoreTokens()) {
      80. 

    80.                  word.set(itr.nextToken());
      81. 

    81.                  context.write(word, one);
      82. 

    82.                  Counter counter = context.getCounter(CountersEnum.class.getName(),
      83. 

    83.                          CountersEnum.INPUT_WORDS.toString());
      84. 

    84.                  counter.increment(1);
      85. 

    85.              }
      86. 

    86.          }
      87. 

    87.      }
      88. 

      89. 

    89.      public static class IntSumReducer
      90. 

    90.              extends Reducer<Text,IntWritable,Text,IntWritable> {
      91. 

    91.          private IntWritable result = new IntWritable();
      92. 

      93. 

    93.          public void reduce(Text key, Iterable<IntWritable> values,
      94. 

    94.                             Context context
      95. 

    95.          ) throws IOException, InterruptedException {
      96. 

    96.              int sum = 0;
      97. 

    97.              for (IntWritable val : values) {
      98. 

    98.                  sum += val.get();
      99. 

    99.              }
      100. 

    100.              result.set(sum);
      101. 

    101.              context.write(key, result);
      102. 

    102.          }
      103. 

    103.      }
      104. 

      105. 

    105.      public static void main(String[] args) throws Exception {
      106. 

    106.          Configuration conf = new Configuration();
      107. 

      108. 

    108.          conf.set("fs.oss.accessKeyId", "${accessKeyId}");
      109. 

    109.         conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
      110. 

    110.          conf.set("fs.oss.endpoint","${endpoint}");
      111. 

      112. 

    112.          GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
      113. 

    113.          String[] remainingArgs = optionParser.getRemainingArgs();
      114. 

    114.          if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
      115. 

    115.              System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
      116. 

    116.              System.exit(2);
      117. 

    117.          }
      118. 

    118.          Job job = Job.getInstance(conf, "word count");
      119. 

    119.          job.setJarByClass(WordCount2.class);
      120. 

    120.          job.setMapperClass(TokenizerMapper.class);
      121. 

    121.          job.setCombinerClass(IntSumReducer.class);
      122. 

    122.          job.setReducerClass(IntSumReducer.class);
      123. 

    123.          job.setOutputKeyClass(Text.class);
      124. 

    124.          job.setOutputValueClass(IntWritable.class);
      125. 

      126. 

    126.          List<String> otherArgs = new ArrayList<String>();
      127. 

    127.          for (int i=0; i < remainingArgs.length; ++i) {
      128. 

    128.              if ("-skip".equals(remainingArgs[i])) {
      129. 

    129.                  job.addCacheFile(new Path(EMapReduceOSSUtil.buildOSSCompleteUri(remainingArgs[++i], conf)).toUri());
      130. 

    130.                  job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
      131. 

    131.              } else {
      132. 

    132.                  otherArgs.add(remainingArgs[i]);
      133. 

    133.              }
      134. 

    134.          }
      135. 

    135.          FileInputFormat.addInputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(0), conf)));
      136. 

    136.          FileOutputFormat.setOutputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(1), conf)));
      137. 

      138. 

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

    139.      }
      140. 

    140.  }
      141.

    其中的 EMapReduceOSSUtil 类代码请参见如下示例，放在和 WordCount2 相同目录：

      

    1.  package com.aliyun.emr.hadoop.examples;
      2. 

      3. 

    3.  import org.apache.hadoop.conf.Configuration;
      4. 

      5. 

    5.  public class EMapReduceOSSUtil {
      6. 

      7. 

    7.      private static String SCHEMA = "oss://";
      8. 

    8.      private static String AKSEP = ":";
      9. 

    9.      private static String BKTSEP = "@";
      10. 

    10.      private static String EPSEP = ".";
      11. 

    11.      private static String HTTP_HEADER = "http://";
      12. 

      13. 

    13.      /**
      14. 

    14.       * complete OSS uri
      15. 

    15.       * convert uri like: oss://bucket/path  to  oss://accessKeyId:aaa@qq.com/path
      16. 

    16.       * ossref do not need this
      17. 

    17.       *
      18. 

    18.       * @param oriUri original OSS uri
      19. 

    19.       */
      20. 

    20.      public static String buildOSSCompleteUri(String oriUri, String akId, String akSecret, String endpoint) {
      21. 

    21.          if (akId == null) {
      22. 

    22.              System.err.println("miss accessKeyId");
      23. 

    23.              return oriUri;
      24. 

    24.          }
      25. 

    25.          if (akSecret == null) {
      26. 

    26.              System.err.println("miss accessKeySecret");
      27. 

    27.              return oriUri;
      28. 

    28.          }
      29. 

    29.          if (endpoint == null) {
      30. 

    30.              System.err.println("miss endpoint");
      31. 

    31.              return oriUri;
      32. 

    32.          }
      33. 

      34. 

    34.          int index = oriUri.indexOf(SCHEMA);
      35. 

    35.          if (index == -1 || index != 0) {
      36. 

    36.              return oriUri;
      37. 

    37.          }
      38. 

      39. 

    39.          int bucketIndex = index + SCHEMA.length();
      40. 

    40.          int pathIndex = oriUri.indexOf("/", bucketIndex);
      41. 

    41.          String bucket = null;
      42. 

    42.          if (pathIndex == -1) {
      43. 

    43.              bucket = oriUri.substring(bucketIndex);
      44. 

    44.          } else {
      45. 

    45.              bucket = oriUri.substring(bucketIndex, pathIndex);
      46. 

    46.          }
      47. 

      48. 

    48.          StringBuilder retUri = new StringBuilder();
      49. 

    49.          retUri.append(SCHEMA)
      50. 

    50.                  .append(akId)
      51. 

    51.                  .append(AKSEP)
      52. 

    52.                  .append(akSecret)
      53. 

    53.                  .append(BKTSEP)
      54. 

    54.                  .append(bucket)
      55. 

    55.                  .append(EPSEP)
      56. 

    56.                  .append(stripHttp(endpoint));
      57. 

      58. 

    58.          if (pathIndex > 0) {
      59. 

    59.              retUri.append(oriUri.substring(pathIndex));
      60. 

    60.          }
      61. 

      62. 

    62.          return retUri.toString();
      63. 

    63.      }
      64. 

      65. 

    65.      public static String buildOSSCompleteUri(String oriUri, Configuration conf) {
      66. 

    66.          return buildOSSCompleteUri(oriUri, conf.get("fs.oss.accessKeyId"), conf.get("fs.oss.accessKeySecret"), conf.get("fs.oss.endpoint"));
      67. 

    67.      }
      68. 

      69. 

    69.      private static String stripHttp(String endpoint) {
      70. 

    70.          if (endpoint.startsWith(HTTP_HEADER)) {
      71. 

    71.              return endpoint.substring(HTTP_HEADER.length());
      72. 

    72.          }
      73. 

    73.          return endpoint;
      74. 

    74.      }
      75. 

    75.  }
      76.

  5. 编译并打包上传。在工程的目录下，执行如下命令：

    1.  mvn clean package -DskipTests

    您即可在工程目录的 target 目录下看到一个 wordcountv2-1.0-SNAPSHOT.jar，这个就是作业 jar 包了。请您将这个 jar 包上传到 OSS 中。

  6. 创建作业。在 E-MapReduce 中新建一个作业，请使用类似如下的参数配置：

    1.  jar ossref://yourBucket/yourPath/wordcountv2-1.0-SNAPSHOT.jar com.aliyun.emr.hadoop.examples.WordCount2 -Dwordcount.case.sensitive=true oss://yourBucket/yourPath/The_Sorrows_of_Young_Werther.txt oss://yourBucket/yourPath/output -skip oss://yourBucket/yourPath/patterns.txt

    这里的 yourBucket 是您的一个 OSS bucket，yourPath 是这个 bucket 上的一个路径，需要您按照实际情况填写。请您将oss://yourBucket/yourPath/The_Sorrows_of_Young_Werther.txt和oss://yourBucket/yourPath/patterns.txt这两个用来处理相关资源的文件下载下来并放到您的 OSS 上。作业需要资源可以从下面下载，然后放到您的 OSS 对应目录下。

    资源下载：The_Sorrows_of_Young_Werther.txtpatterns.txt

  7. 创建执行计划并运行。在 E-MapReduce 中创建执行计划，关联这个作业并运行。

上一篇： Android ExpandableListView的使用

下一篇： 汉诺塔问题（递归思想）

推荐阅读