Create an EMR MR node

Prepare the sample data

Create a sample file named input01.txt with the following content.

hadoop emr hadoop dw
hive hadoop
dw emr

Upload the sample data file

1. Log on to the OSS console. In the left-side navigation pane, click Buckets.

2. Click the name of the target bucket to open the File Management page.

   This topic uses a bucket named onaliyun-bucket-2 as an example.

3. Click New Directory to create directories for the sample data and JAR resources.

   • Set Directory Name to emr/datas/wordcount02/inputs to create a directory for the sample data.

   • Set Directory Name to emr/jars to create a directory for the JAR resources.

4. Upload the sample data file to the data directory.

   • Go to the /emr/datas/wordcount02/inputs path and click Upload File.

   • In the Files to Upload area, click Select Files, add the input01.txt file to the bucket, and then click Upload File.

Build a MapReduce JAR package

1. Open your IntelliJ IDEA project and add the following POM dependencies.

           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-mapreduce-client-common</artifactId>
               <version>2.8.5</version> <!-- EMR MR uses version 2.8.5. -->
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>2.8.5</version>
           </dependency>

2. To read from and write to OSS files in MapReduce, you must configure the following parameters.

   Important

   Risk warning: Your Alibaba Cloud account's AccessKey has full access to all APIs. We recommend using a RAM user for API access and daily O&M. Do not store your AccessKey ID and AccessKey Secret in your project code or anywhere they might be exposed. A leaked AccessKey can compromise the security of all resources under your account. The following code sample is for reference only. Keep your AccessKey information secure.

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

   The following table describes the parameters.

   • ${accessKeyId}: The AccessKey ID of your Alibaba Cloud account.

   • ${accessKeySecret}: The AccessKey secret of your Alibaba Cloud account.

   • ${endpoint}: The OSS endpoint. The OSS bucket must be in the same region as the EMR cluster, and this region determines the endpoint. For more information, see Regions and endpoints.

   The following Java code provides an example of how to modify the WordCount example from the official Hadoop website. It adds the AccessKey ID and AccessKey secret to grant the job permission to access the OSS file.

   Sample code

   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. Package the code into a JAR file.

   After you edit and save the Java code, package it into a JAR file. The JAR package generated in this example is named onaliyun_mr_wordcount-1.0-SNAPSHOT.jar.