DataWorks:Create an EMR MR node

Prepare initial data

Create a file that is named input01.txt and contains the following initial data:

hadoop emr hadoop dw
hive hadoop
dw emr

Upload the file that stores the initial data

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

2. On the Buckets page, find the desired bucket and click the bucket name to go to the Objects page.

   In this example, the onaliyun-bucket-2 bucket is used.

3. On the Objects page, click Create Directory to create directories that are used to store initial data and JAR resources.

   • Set Directory Name to emr/datas/wordcount02/inputs to create a directory that is used to store initial data.

   • Set Directory Name to emr/jars to create a directory that is used to store JAR resources.

4. Upload the file that stores the initial data to the emr/datas/wordcount02/inputs directory.

   • Go to the /emr/datas/wordcount02/inputs directory and click Upload Object.

   • In the Files to Upload section, click Select Files and upload the input01.txt file to the bucket.

Use the EMR MR node to read the OSS object and generate a JAR package

1. Open an existing IntelliJ IDEA project and add Project Object Model (POM) dependencies.

           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-mapreduce-client-common</artifactId>
               <version>2.8.5</version> <!--The version used by EMR MR is 2.8.5.-->
           </dependency>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>2.8.5</version>
           </dependency>

2. Configure the following parameters to read data from and write data to the OSS object.

   Important

   An Alibaba Cloud account has permissions to call all API operations. If the AccessKey pair of your Alibaba Cloud account is leaked, all resources in your Alibaba Cloud account may be exposed to high security risks. We recommend that you do not save the AccessKey ID and AccessKey secret of your Alibaba Cloud account into the project code or positions that can be easily located. We recommend that you use a RAM user to call API operations or perform routine O&M. The following sample code is provided only for reference. Keep the AccessKey pair of your Alibaba Cloud account confidential.

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

   Parameter description:

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

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

   • ${endpoint}: the endpoint of OSS. The endpoint is determined by the region where your EMR cluster resides. You must activate OSS in the region where your EMR cluster resides. For more information, see OSO regions and endpoints.

   In this topic, the Java code is used to modify the WordCount example on the Hadoop official website. The configuration of the AccessKey ID and AccessKey secret is added to the code. This grants the job the permissions to access OSS objects.

   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 write and save the preceding code, compress the code into a JAR package. In this example, a package named onaliyun_mr_wordcount-1.0-SNAPSHOT.jar is generated.