DataWorks:EMR MR Node

Prerequisites

Before you begin, make sure you have:

• An EMR cluster created and registered with DataWorks. See Data Studio: Associate an EMR computing resource

• (Optional) If you use a Resource Access Management (RAM) user, add the user to the workspace with the Developer or Workspace Administrator role. The Workspace Administrator role has extensive permissions — grant it with caution. See Add members to a workspace. Alibaba Cloud account users can skip this step

• An OSS bucket. See Create buckets

Limitations

• EMR MR nodes can only run on a Serverless Resource Group (recommended) or an Exclusive Scheduling Resource Group.

• For DataLake or custom clusters, configure EMR-HOOK on the cluster before managing metadata in DataWorks. See Configure Hive EMR-HOOK.

  Note

  Without EMR-HOOK, DataWorks cannot display metadata in real time, generate audit logs, display Data Lineage, or perform EMR-related data governance tasks.

How it works

This tutorial has five stages:

Step 1: Prepare source data

hadoop emr hadoop dw
hive hadoop
dw emr

Step 2: Build the JAR package

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

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);
    }
}

Step 3: Configure the EMR MR node

Data Studio provides two ways to make the JAR available to the EMR MR node. Choose based on your situation:

Step 4: Run the job

1. In the Run Configuration section, set Compute Resources and Resource Group.

   Note

   • You can also configure Scheduling CUs based on the resource requirements of the task. The default CU value is 0.25.

   • To access data sources over the public internet or a VPC, use a scheduling resource group that has passed the connectivity test with the data source. See Network connectivity solutions.

2. In the parameter dialog in the toolbar, select the data source and click Run.

3. To run the node on a schedule, configure scheduling properties. See Node scheduling configuration.

4. Deploy the node to make it operational. See Node and workflow deployment.

Step 5: View results

After the job completes, check the output in two ways:

In the OSS console: Navigate to the emr/datas/wordcount02/outputs directory in onaliyun-bucket-2 to find the output files.

In DataWorks using Hive: Create an EMR Hive node (see Create a node for a scheduled workflow), then create an external Hive table that points to the output path and query it:

CREATE EXTERNAL TABLE IF NOT EXISTS wordcount02_result_tb
(
    `word` STRING COMMENT 'word',
    `cout` 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 query returns the word count results stored in the output directory.

What's next

• Monitor the job's running status in Operation Center. See Getting started with Operation Center.

• Schedule the node to run periodically. See Node scheduling configuration.

• Explore other EMR node types in Data Studio, such as EMR Hive or EMR Spark, for different workload patterns.