本文介绍如何在E-MapReduce集群中开发MapReduce作业流程。
在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也需要是在集群对应的地域，详情请参见访问域名和数据中心。
Word Count
以下示例介绍了如何从OSS中读取文本，然后统计其中单词的数量。其操作步骤如下：
- 程序编写
以Java代码为例，修改Hadoop官网WordCount例子，即在代码中添加AccessKey ID和AccessKey Secret的配置，以便作业有权限访问OSS文件。
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); } } - 编译程序
配置好JDK和Hadoop环境后，执行如下命令。
mkdir wordcount_classes 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 jar cvf wordcount.jar -C wordcount_classesHADOOP_HOME：通常Hadoop的安装目录为/usr/lib/hadoop-current。您也可以通过SSH方式登录Master节点，执行env |grep hadoop命令获取，详情请参见使用SSH连接主节点。 - 创建作业
- 将步骤2的JAR文件上传到OSS，详情请参见上传文件。
例如，JAR文件在OSS上的路径为oss://emr/jars/wordcount.jar，输入输出路径分别为oss://emr/data/WordCount/Input和oss://emr/data/WordCount/Output。
- 在E-MapReduce中创建作业，详情请参见Hadoop MapReduce作业配置。
- 将步骤2的JAR文件上传到OSS，详情请参见上传文件。
- 运行作业
在作业编辑中，单击运行。
使用Maven工程来管理MR作业
- 安装Maven
首先确保您已经安装了Maven。
- 生成工程框架
在您的工程根目录处（例如，您的工程开发根目录是D:/workspace）执行如下命令。
mvn archetype:generate -DgroupId=com.aliyun.emr.hadoop.examples -DartifactId=wordcountv2 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=falsemvn会自动生成一个空的Sample工程位于D:/workspace/wordcountv2（和您指定的artifactId一致），里面包含一个简单的pom.xml和App类（类的包路径和您指定的groupId一致）。
- 加入Hadoop依赖
使用IDE打开Sample工程，编辑pom.xml文件，即在
<dependency>中添加如下内容。<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-common</artifactId> <version>2.6.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.6.0</version> </dependency> - 编写代码
在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); } }其中EMapReduceOSSUtil类代码请参见如下示例，放在和WordCount2相同的目录下。package com.aliyun.emr.hadoop.examples; import org.apache.hadoop.conf.Configuration; public class EMapReduceOSSUtil { private static String SCHEMA = "oss://"; private static String AKSEP = ":"; private static String BKTSEP = "@"; private static String EPSEP = "."; private static String HTTP_HEADER = "http://"; /** * complete OSS uri * convert uri like: oss://bucket/path to oss://accessKeyId:accessKeySecret@bucket.endpoint/path * ossref do not need this * * @param oriUri original OSS uri */ public static String buildOSSCompleteUri(String oriUri, String akId, String akSecret, String endpoint) { if (akId == null) { System.err.println("miss accessKeyId"); return oriUri; } if (akSecret == null) { System.err.println("miss accessKeySecret"); return oriUri; } 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(akId) .append(AKSEP) .append(akSecret) .append(BKTSEP) .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.accessKeyId"), conf.get("fs.oss.accessKeySecret"), conf.get("fs.oss.endpoint")); } private static String stripHttp(String endpoint) { if (endpoint.startsWith(HTTP_HEADER)) { return endpoint.substring(HTTP_HEADER.length()); } return endpoint; } } - 编译并打包上传
在工程的目录下，执行如下命令。
mvn clean package -DskipTests您可以在工程目录的target目录下看到wordcountv2-1.0-SNAPSHOT.jar，请上传JAR包到OSS中。
- 创建作业
在E-MapReduce中新建一个作业，作业内容如下所示。
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://<yourBucketName>/<yourPath>/output -skip oss://<yourBucketName>/<yourPath>/patterns.txt其中
<yourBucketName>是OSS Bucket的名称，<yourPath>是OSS Bucket上的路径，需要您按照实际情况填写。请下载oss://<yourBucketName>/<yourPath>/The_Sorrows_of_Young_Werther.txt和oss://<yourBucketName>/<yourPath>/patterns.txt并上传至您的OSS上。您可以下载作业所需资源并上传至您OSS对应目录下： - 运行作业
在作业编辑中，单击运行。
在文档使用中是否遇到以下问题
更多建议
匿名提交