DataWorks:Create an EMR MR node

Background information

• You must replace the object name, bucket name, and directory that are involved in the development process with the actual information.

• You can use the OSS REF method to reference OSS resources. For more information, see the Method 1: Reference an OSS resource section in this topic.

Limits

• Tasks on this type of node cannot be run on the shared resource group for scheduling. You can run the tasks only on exclusive resource groups that are created after December 1, 2023.

• If you want to manage metadata for a DataLake or custom cluster in DataWorks, you must configure EMR-HOOK in the cluster first. If you do not configure EMR-HOOK in the cluster, metadata cannot be displayed in real time, audit logs cannot be generated, and data lineages cannot be displayed in DataWorks. In addition, the related EMR governance tasks cannot be run. For more information about how to configure EMR-HOOK, see Use the Hive extension feature to record data lineage and historical access information.

Prepare initial data and a JAR resource package

1. 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

2. Create directories that are used to store initial data and JAR resources.

   1. Log on to the OSS console.

   2. In the left-side navigation pane, click Buckets.

   3. 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.

   4. 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.

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

      1. Go to the emr/datas/wordcount02/inputs directory.

      2. Click Upload.

      3. In the Files to Upload section, click Select Files and add the input01.txt file.

      4. Click Upload.

3. 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 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.

      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. After you write 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.

Create an EMR MR node

1. Go to the DataStudio page.

   Log on to the DataWorks console. In the left-side navigation pane, choose Data Modeling and Development > DataStudio. On the page that appears, select the desired workspace from the drop-down list and click Go to DataStudio.

2. Go to the DataStudio page.

   Log on to the DataWorks console. In the left-side navigation pane, choose Data Modeling and Development > DataStudio. On the page that appears, select the desired workspace from the drop-down list and click Go to DataStudio.

3. Create an EMR MR node.

   1. Move the pointer over the icon and choose Create Node > EMR > EMR MR.

      Alternatively, you can find the desired workflow, right-click the name of the workflow, and then choose Create Node > EMR > EMR MR.

   2. In the Create Node dialog box, configure the Name, Engine Instance, Node Type, and Path parameters.

      Note

      The node name must be 1 to 128 characters in length and can contain only letters, digits, underscores (_), and periods (.).

   3. Click Confirm. The configuration tab of the EMR MR node appears.

Develop an MR task

Use one of the following methods to develop an MR task on the configuration tab of the EMR MR node:

ossref://{endpoint}/{bucket}/{object}

Configure scheduling properties

If you want the system to periodically run the node, you can click Properties in the right-side navigation pane on the configuration tab of the node to configure properties for the node based on your business requirements.

• Configure basic properties for the node. For more information, see Configure basic properties.

• Configure the scheduling cycle, rerun properties, and scheduling dependencies of the node. For more information, see Configure time properties and Configure same-cycle scheduling dependencies.

  Note

  Before you commit the node, you must configure the Rerun and Parent Nodes parameters on the Properties tab.

• Configure resource properties for the node. For more information, see Configure the resource property. If the node that you created is an auto triggered node and you want the node to access the Internet or a virtual private cloud (VPC), you must select the resource group for scheduling that is connected to the node. For more information, see Establish a network connection between a resource group and a data source.

Deploy the node for periodic scheduling

• Commit and deploy the node.

  1. Click the icon in the top toolbar to save the node.

  2. Click the icon in the top toolbar to commit the node.

  3. In the Commit Node dialog box, configure the Change description parameter.

  4. Click OK.

  If you use a workspace in standard mode, you must deploy the node in the production environment after you commit the node. On the left side of the top navigation bar, click Deploy. For more information, see Deploy nodes.

• View the node.

  1. Click Operation Center in the upper-right corner of the configuration tab of the node to go to Operation Center in the production environment.

  2. View the scheduled node. For more information, see View and manage auto triggered nodes.

  To view more information about the node, click Operation Center in the top navigation bar of the DataStudio page. For more information, see Overview.

View the results

• Log on to the OSS console. Then, you can view the results in the emr/datas/wordcount02/outputs directory in which the initial data is stored.

• View the statistical results in the DataWorks console.

  1. Create an EMR Hive node. For more information, see Create an EMR Hive node.

  2. On the EMR Hive node, create a Hive external table that is mounted to OSS. Then, use the Hive external table to read data from Hive tables in OSS. Sample code:

     CREATE EXTERNAL TABLE IF NOT EXISTS wordcount02_result_tb
     (
         `word` STRING COMMENT 'Word',
         `count` STRING COMMENT 'Count'   
     ) 
     ROW FORMAT delimited fields terminated by '\t'
     location 'oss://onaliyun-bucket-2/emr/datas/wordcount02/outputs/';
     
     SELECT * FROM wordcount02_result_tb;

     The following figure shows the results.