全部产品
E-MapReduce

MapReduce 开发手册

更新时间:2017-06-07 13:26:11   分享:   

在 MapReduce 中使用 OSS

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

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

参数说明:

${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. import java.io.IOException;
    3. import java.util.StringTokenizer;
    4. import org.apache.hadoop.conf.Configuration;
    5. import org.apache.hadoop.fs.Path;
    6. import org.apache.hadoop.io.IntWritable;
    7. import org.apache.hadoop.io.Text;
    8. import org.apache.hadoop.mapreduce.Job;
    9. import org.apache.hadoop.mapreduce.Mapper;
    10. import org.apache.hadoop.mapreduce.Reducer;
    11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    13. import org.apache.hadoop.util.GenericOptionsParser;
    14. public class EmrWordCount {
    15. public static class TokenizerMapper
    16. extends Mapper<Object, Text, Text, IntWritable>{
    17. private final static IntWritable one = new IntWritable(1);
    18. private Text word = new Text();
    19. public void map(Object key, Text value, Context context
    20. ) throws IOException, InterruptedException {
    21. StringTokenizer itr = new StringTokenizer(value.toString());
    22. while (itr.hasMoreTokens()) {
    23. word.set(itr.nextToken());
    24. context.write(word, one);
    25. }
    26. }
    27. }
    28. public static class IntSumReducer
    29. extends Reducer<Text,IntWritable,Text,IntWritable> {
    30. private IntWritable result = new IntWritable();
    31. public void reduce(Text key, Iterable<IntWritable> values,
    32. Context context
    33. ) throws IOException, InterruptedException {
    34. int sum = 0;
    35. for (IntWritable val : values) {
    36. sum += val.get();
    37. }
    38. result.set(sum);
    39. context.write(key, result);
    40. }
    41. }
    42. public static void main(String[] args) throws Exception {
    43. Configuration conf = new Configuration();
    44. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    45. if (otherArgs.length < 2) {
    46. System.err.println("Usage: wordcount <in> [<in>...] <out>");
    47. System.exit(2);
    48. }
    49. conf.set("fs.oss.accessKeyId", "${accessKeyId}");
    50. conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
    51. conf.set("fs.oss.endpoint","${endpoint}");
    52. Job job = Job.getInstance(conf, "word count");
    53. job.setJarByClass(EmrWordCount.class);
    54. job.setMapperClass(TokenizerMapper.class);
    55. job.setCombinerClass(IntSumReducer.class);
    56. job.setReducerClass(IntSumReducer.class);
    57. job.setOutputKeyClass(Text.class);
    58. job.setOutputValueClass(IntWritable.class);
    59. for (int i = 0; i < otherArgs.length - 1; ++i) {
    60. FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
    61. }
    62. FileOutputFormat.setOutputPath(job,
    63. new Path(otherArgs[otherArgs.length - 1]));
    64. System.exit(job.waitForCompletion(true) ? 0 : 1);
    65. }
    66. }
  2. 编译程序。首先要将 jdk 和 Hadoop 环境配置好,然后进行如下操作:

    1. mkdir wordcount_classes
    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. jar cvf wordcount.jar -C wordcount_classes .
  3. 创建作业。

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

    • E-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. <groupId>org.apache.hadoop</groupId>
    3. <artifactId>hadoop-mapreduce-client-common</artifactId>
    4. <version>2.6.0</version>
    5. </dependency>
    6. <dependency>
    7. <groupId>org.apache.hadoop</groupId>
    8. <artifactId>hadoop-common</artifactId>
    9. <version>2.6.0</version>
    10. </dependency>
  4. 编写代码。在 com.aliyun.emr.hadoop.examples 包下和 App 类平行的位置添加新类 WordCount2.java。内容如下:

    1. package com.aliyun.emr.hadoop.examples;
    2. import java.io.BufferedReader;
    3. import java.io.FileReader;
    4. import java.io.IOException;
    5. import java.net.URI;
    6. import java.util.ArrayList;
    7. import java.util.HashSet;
    8. import java.util.List;
    9. import java.util.Set;
    10. import java.util.StringTokenizer;
    11. import org.apache.hadoop.conf.Configuration;
    12. import org.apache.hadoop.fs.Path;
    13. import org.apache.hadoop.io.IntWritable;
    14. import org.apache.hadoop.io.Text;
    15. import org.apache.hadoop.mapreduce.Job;
    16. import org.apache.hadoop.mapreduce.Mapper;
    17. import org.apache.hadoop.mapreduce.Reducer;
    18. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    19. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    20. import org.apache.hadoop.mapreduce.Counter;
    21. import org.apache.hadoop.util.GenericOptionsParser;
    22. import org.apache.hadoop.util.StringUtils;
    23. public class WordCount2 {
    24. public static class TokenizerMapper
    25. extends Mapper<Object, Text, Text, IntWritable>{
    26. static enum CountersEnum { INPUT_WORDS }
    27. private final static IntWritable one = new IntWritable(1);
    28. private Text word = new Text();
    29. private boolean caseSensitive;
    30. private Set<String> patternsToSkip = new HashSet<String>();
    31. private Configuration conf;
    32. private BufferedReader fis;
    33. @Override
    34. public void setup(Context context) throws IOException,
    35. InterruptedException {
    36. conf = context.getConfiguration();
    37. caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
    38. if (conf.getBoolean("wordcount.skip.patterns", true)) {
    39. URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
    40. for (URI patternsURI : patternsURIs) {
    41. Path patternsPath = new Path(patternsURI.getPath());
    42. String patternsFileName = patternsPath.getName().toString();
    43. parseSkipFile(patternsFileName);
    44. }
    45. }
    46. }
    47. private void parseSkipFile(String fileName) {
    48. try {
    49. fis = new BufferedReader(new FileReader(fileName));
    50. String pattern = null;
    51. while ((pattern = fis.readLine()) != null) {
    52. patternsToSkip.add(pattern);
    53. }
    54. } catch (IOException ioe) {
    55. System.err.println("Caught exception while parsing the cached file '"
    56. + StringUtils.stringifyException(ioe));
    57. }
    58. }
    59. @Override
    60. public void map(Object key, Text value, Context context
    61. ) throws IOException, InterruptedException {
    62. String line = (caseSensitive) ?
    63. value.toString() : value.toString().toLowerCase();
    64. for (String pattern : patternsToSkip) {
    65. line = line.replaceAll(pattern, "");
    66. }
    67. StringTokenizer itr = new StringTokenizer(line);
    68. while (itr.hasMoreTokens()) {
    69. word.set(itr.nextToken());
    70. context.write(word, one);
    71. Counter counter = context.getCounter(CountersEnum.class.getName(),
    72. CountersEnum.INPUT_WORDS.toString());
    73. counter.increment(1);
    74. }
    75. }
    76. }
    77. public static class IntSumReducer
    78. extends Reducer<Text,IntWritable,Text,IntWritable> {
    79. private IntWritable result = new IntWritable();
    80. public void reduce(Text key, Iterable<IntWritable> values,
    81. Context context
    82. ) throws IOException, InterruptedException {
    83. int sum = 0;
    84. for (IntWritable val : values) {
    85. sum += val.get();
    86. }
    87. result.set(sum);
    88. context.write(key, result);
    89. }
    90. }
    91. public static void main(String[] args) throws Exception {
    92. Configuration conf = new Configuration();
    93. conf.set("fs.oss.accessKeyId", "${accessKeyId}");
    94. conf.set("fs.oss.accessKeySecret", "${accessKeySecret}");
    95. conf.set("fs.oss.endpoint","${endpoint}");
    96. GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    97. String[] remainingArgs = optionParser.getRemainingArgs();
    98. if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
    99. System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
    100. System.exit(2);
    101. }
    102. Job job = Job.getInstance(conf, "word count");
    103. job.setJarByClass(WordCount2.class);
    104. job.setMapperClass(TokenizerMapper.class);
    105. job.setCombinerClass(IntSumReducer.class);
    106. job.setReducerClass(IntSumReducer.class);
    107. job.setOutputKeyClass(Text.class);
    108. job.setOutputValueClass(IntWritable.class);
    109. List<String> otherArgs = new ArrayList<String>();
    110. for (int i=0; i < remainingArgs.length; ++i) {
    111. if ("-skip".equals(remainingArgs[i])) {
    112. job.addCacheFile(new Path(EMapReduceOSSUtil.buildOSSCompleteUri(remainingArgs[++i], conf)).toUri());
    113. job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
    114. } else {
    115. otherArgs.add(remainingArgs[i]);
    116. }
    117. }
    118. FileInputFormat.addInputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(0), conf)));
    119. FileOutputFormat.setOutputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(1), conf)));
    120. System.exit(job.waitForCompletion(true) ? 0 : 1);
    121. }
    122. }

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

    1. package com.aliyun.emr.hadoop.examples;
    2. import org.apache.hadoop.conf.Configuration;
    3. public class EMapReduceOSSUtil {
    4. private static String SCHEMA = "oss://";
    5. private static String AKSEP = ":";
    6. private static String BKTSEP = "@";
    7. private static String EPSEP = ".";
    8. private static String HTTP_HEADER = "http://";
    9. /**
    10. * complete OSS uri
    11. * convert uri like: oss://bucket/path to oss://accessKeyId:accessKeySecret@bucket.endpoint/path
    12. * ossref do not need this
    13. *
    14. * @param oriUri original OSS uri
    15. */
    16. public static String buildOSSCompleteUri(String oriUri, String akId, String akSecret, String endpoint) {
    17. if (akId == null) {
    18. System.err.println("miss accessKeyId");
    19. return oriUri;
    20. }
    21. if (akSecret == null) {
    22. System.err.println("miss accessKeySecret");
    23. return oriUri;
    24. }
    25. if (endpoint == null) {
    26. System.err.println("miss endpoint");
    27. return oriUri;
    28. }
    29. int index = oriUri.indexOf(SCHEMA);
    30. if (index == -1 || index != 0) {
    31. return oriUri;
    32. }
    33. int bucketIndex = index + SCHEMA.length();
    34. int pathIndex = oriUri.indexOf("/", bucketIndex);
    35. String bucket = null;
    36. if (pathIndex == -1) {
    37. bucket = oriUri.substring(bucketIndex);
    38. } else {
    39. bucket = oriUri.substring(bucketIndex, pathIndex);
    40. }
    41. StringBuilder retUri = new StringBuilder();
    42. retUri.append(SCHEMA)
    43. .append(akId)
    44. .append(AKSEP)
    45. .append(akSecret)
    46. .append(BKTSEP)
    47. .append(bucket)
    48. .append(EPSEP)
    49. .append(stripHttp(endpoint));
    50. if (pathIndex > 0) {
    51. retUri.append(oriUri.substring(pathIndex));
    52. }
    53. return retUri.toString();
    54. }
    55. public static String buildOSSCompleteUri(String oriUri, Configuration conf) {
    56. return buildOSSCompleteUri(oriUri, conf.get("fs.oss.accessKeyId"), conf.get("fs.oss.accessKeySecret"), conf.get("fs.oss.endpoint"));
    57. }
    58. private static String stripHttp(String endpoint) {
    59. if (endpoint.startsWith(HTTP_HEADER)) {
    60. return endpoint.substring(HTTP_HEADER.length());
    61. }
    62. return endpoint;
    63. }
    64. }
  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.txtoss://yourBucket/yourPath/patterns.txt这两个用来处理相关资源的文件下载下来并放到您的 OSS 上。作业需要资源可以从下面下载,然后放到您的 OSS 对应目录下。

    资源下载:The_Sorrows_of_Young_Werther.txtpatterns.txt

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

本文导读目录
本文导读目录
以上内容是否对您有帮助?