创建EMR MR节点

准备初始数据

创建示例文件input01.txt,文件内容如下。

hadoop emr hadoop dw
hive hadoop
dw emr

上传初始数据文件

1. 登录OSS管理控制台，单击左侧导航栏的Bucket列表。

2. 单击目标Bucket名称，进入文件管理页面。

   本文示例使用的Bucket为onaliyun-bucket-2。

3. 单击新建目录，创建初始数据及JAR资源的存放目录。

   • 配置目录名为emr/datas/wordcount02/inputs，创建初始数据的存放目录。

   • 配置目录名为emr/jars，创建JAR资源的存放目录。

4. 上传初始数据文件至初始数据的存放目录。

   • 进入/emr/datas/wordcount02/inputs路径，单击上传文件。

   • 在待上传文件区域单击扫描文件，添加input01.txt文件至Bucket，单击上传文件。

使用MapReduce读取OSS文件并生成JAR包

1. 打开已创建的IDEA项目，添加pom依赖。

           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-mapreduce-client-common</artifactId>
               <version>2.8.5</version> <!--因为EMR-MR用的是2.8.5-->
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>2.8.5</version>
           </dependency>

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

   重要

   风险提示： 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。强烈建议不要将AccessKey ID和AccessKey Secret保存到工程代码里或者任何容易被泄露的地方，AccessKey泄露会威胁您账号下所有资源的安全。以下代码示例仅供参考，请妥善保管好您的AccessKey信息。

   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也需要是在集群对应的地域，详情请参见OSS地域和访问域名

   以Java代码为例，修改Hadoop官网WordCount示例，即在代码中添加AccessKey ID和AccessKey Secret的配置，以便作业有权限访问OSS文件。

   示例代码

   package cn.apache.hadoop.onaliyun.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);
       }
   }          

3. 将代码打包成JAR文件。

   编辑保存Java代码后，将Java代码打包成JAR文件。示例生成的JAR包为onaliyun_mr_wordcount-1.0-SNAPSHOT.jar。