Nó EMR MR - DataWorks: Central de Documentação do Alibaba Cloud

No DataWorks, você pode criar um nó E-MapReduce (EMR) MapReduce (MR) para dividir grandes conjuntos de dados em tarefas de mapeamento paralelas, o que aumenta significativamente a eficiência do processamento de dados. Este tópico apresenta um exemplo de desenvolvimento e configuração de um job EMR MR que lê um arquivo de texto do Object Storage Service (OSS) e conta as palavras contidas nele.

Pré-requisitos

crie um cluster EMR do Alibaba Cloud e registre-o no DataWorks. Para mais informações, consulte New Data Studio: Attach an EMR compute resource.
(Opcional, para usuários do RAM) O usuário do Resource Access Management (RAM) responsável pelo desenvolvimento da tarefa deve ser adicionado ao workspace e receber a função Development ou Workspace Administrator (esta função inclui permissões amplas e deve ser concedida com cautela). Para mais informações, consulte Add workspace members.

Se você estiver usando uma conta raiz, pule esta etapa.
Para acompanhar o exemplo deste tópico, crie um bucket no Object Storage Service (OSS). Para mais informações, consulte Create buckets.

Limitações

execute este tipo de nó apenas em um serverless resource group (recomendado) ou em um grupo de recursos exclusivo para agendamento.
gerencie os metadados de um cluster DataLake ou personalizado no DataWorks, configure primeiro o EMR-HOOK no cluster. Para mais informações, consulte Configure Hive EMR-HOOK.

Nota
Se o EMR-HOOK não estiver configurado no cluster, o DataWorks não poderá exibir metadados em tempo real, gerar logs de auditoria, exibir linhagem de dados ou executar tarefas de governança de dados relacionadas ao EMR.

Prepare os dados iniciais e o pacote JAR

Prepare os dados iniciais

crie um arquivo de exemplo chamado input01.txt com o seguinte conteúdo.

hadoop emr hadoop dw
hive hadoop
dw emr

Faça o upload do arquivo de dados inicial

acesse o console do OSS. No painel de navegação à esquerda, clique em Buckets.
Clique no nome do bucket de destino para abrir a página File Management.

Este exemplo usa um bucket chamado onaliyun-bucket-2.
Clique em Create Directory para criar diretórios para os dados iniciais e o recurso JAR.
- defina Directory Name como emr/datas/wordcount02/inputs para criar o diretório dos dados iniciais.
- defina Directory Name como emr/jars para criar o diretório do recurso JAR.
Faça o upload do arquivo de dados iniciais para o seu diretório.
- acesse o caminho /emr/datas/wordcount02/inputs e clique em Upload File.
- Na área Files to Upload, clique em Select Files, adicione o arquivo input01.txt ao bucket e, em seguida, clique em Upload File.

Crie um job MapReduce e o pacote JAR

Abra seu projeto no IntelliJ IDEA e adicione as seguintes dependências ao arquivo pom.xml.

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

Para ler e gravar em um arquivo do OSS no MapReduce, configure os seguintes parâmetros.

Importante

Aviso de risco: O par de AccessKey da sua conta Alibaba Cloud concede acesso total a todas as operações de api. Expor seu AccessKey id e AccessKey Secret compromete a segurança de todos os recursos da sua conta. Recomendamos fortemente o uso de um usuário RAM para chamadas de API e operações diárias. Não codifique seu AccessKey ID ou AccessKey Secret diretamente no código do seu projeto ou em qualquer outro local acessível publicamente. O código a seguir é apenas para fins de demonstração. Mantenha suas informações de AccessKey seguras.

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

A tabela a seguir descreve os parâmetros.

${accessKeyId}: O AccessKey id da sua conta Alibaba Cloud.
${accessKeySecret}: O AccessKey Secret da sua conta Alibaba Cloud.
${endpoint}: O endpoint público do OSS. O endpoint depende da região onde o seu cluster EMR está localizado. O bucket do OSS e o cluster devem estar na mesma região. Para mais informações, consulte Regions and endpoints.

O código Java a seguir é uma versão modificada do exemplo oficial Hadoop WordCount. Ele inclui configurações para o AccessKey id e o AccessKey Secret para conceder ao job permissão de acesso aos arquivos do OSS.

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

edite o código Java e empacote-o em um arquivo JAR. Este exemplo gera um arquivo JAR chamado onaliyun_mr_wordcount-1.0-SNAPSHOT.jar.

Procedimento

Na guia de configuração do nó EMR MR, desenvolva a tarefa da seguinte forma:

Desenvolva a tarefa EMR MR

escolha um dos seguintes métodos de acordo com suas necessidades:

Método 1: Faça o upload e referencie o JAR

Você também pode fazer o upload de um recurso da sua máquina local para o DataStudio e depois referenciá-lo em um nó. Se um recurso for muito grande para upload pelo console do DataWorks, armazene-o no HDFS e referencie-o no seu código.

Crie um recurso JAR.
1. Para mais informações, consulte Manage resources. Armazene o pacote JAR da etapa Prepare initial data and a JAR package no diretório emr/jars. Clique em Click Upload.
2. Configure os parâmetros Storage Path, Data Sources e Resource Group.
3. Clique em Save.
Referencie o recurso JAR.
1. Abra o nó EMR MR para acessar sua guia de configuração.
2. No painel Resource Management à esquerda, localize o recurso que deseja referenciar. Neste exemplo, o recurso é onaliyun_mr_wordcount-1.0-SNAPSHOT.jar. Clique com o botão direito no recurso e selecione Insert Resource Path.
3. Após referenciar o recurso, uma instrução de referência aparece na guia de configuração do nó EMR MR, indicando que o recurso foi referenciado com sucesso. Execute o seguinte comando. Substitua o pacote de recursos, o nome do bucket e o caminho pelas suas informações reais.
```
##@resource_reference{"onaliyun_mr_wordcount-1.0-SNAPSHOT.jar"}
onaliyun_mr_wordcount-1.0-SNAPSHOT.jar cn.apache.hadoop.onaliyun.examples.EmrWordCount oss://onaliyun-bucket-2/emr/datas/wordcount02/inputs oss://onaliyun-bucket-2/emr/datas/wordcount02/outputs
```
  Nota
  Não há suporte para comentários no editor de código para nós EMR MR.

Método 2: Referencie o recurso do OSS

Use o método OSS REF para referenciar diretamente um recurso do OSS. Quando o nó é executado, o DataWorks baixa automaticamente o recurso do OSS referenciado para o ambiente local. Este método é bastante utilizado em cenários onde uma tarefa EMR depende de um arquivo JAR ou de um script.

Faça o upload do recurso JAR.
1. Após desenvolver o código, acesse o console do OSS. No painel de navegação à esquerda, clique em Buckets.
2. Clique no nome do bucket de destino para abrir a página File Management.
  
  Este exemplo usa um bucket chamado onaliyun-bucket-2.
3. Faça o upload do recurso JAR para o seu diretório.
  
  Acesse o diretório emr/jars. Clique em Upload File. Na área Files to Upload, clique em Select Files, adicione o arquivo onaliyun_mr_wordcount-1.0-SNAPSHOT.jar e, em seguida, clique em Upload File.

Referencie o recurso JAR.

Na guia de configuração do nó EMR MR, escreva o código para referenciar o recurso JAR.

hadoop jar ossref://onaliyun-bucket-2/emr/jars/onaliyun_mr_wordcount-1.0-SNAPSHOT.jar cn.apache.hadoop.onaliyun.examples.EmrWordCount oss://onaliyun-bucket-2/emr/datas/wordcount02/inputs oss://onaliyun-bucket-2/emr/datas/wordcount02/outputs

Nota

O comando está no seguinte formato: hadoop jar <Path of the JAR to run> <Full name of the main class> <Input file directory> <Output directory>.

A tabela a seguir descreve os parâmetros para o caminho do JAR.

Parâmetro

Descrição

Caminho do JAR a ser executado

O formato é ossref://{endpoint}/{bucket}/{object}

Endpoint: O endpoint público do OSS. Se este parâmetro for deixado em branco, você só poderá referenciar recursos de um bucket na mesma região do cluster EMR.
Bucket: Um contêiner que o OSS usa para armazenar objetos. Cada Bucket tem um nome exclusivo. Acesse o console do OSS para visualizar todos os Buckets da sua conta.
object: Um objeto específico, que pode ser um arquivo ou um caminho, armazenado em um bucket.

(Opcional) Configure parâmetros avançados

No painel à direita, clique na guia Scheduling Settings. configure os seguintes parâmetros na seção EMR Node Parameters > DataWorks parameters.

Nota

Os parâmetros avançados disponíveis variam de acordo com o tipo de cluster EMR, conforme mostrado nas tabelas a seguir.
configure propriedades adicionais do Spark de código aberto na guia Scheduling Settings, na seção EMR Node Parameters > Spark parameter.

Cluster DataLake e personalizado: EMR no ECS

Parâmetro	Descrição
queue	A fila onde os jobs são enviados. O valor padrão é default. Para mais informações sobre o EMR YARN, consulte Basic queue configurations.
priority	A prioridade do job. O valor padrão é 1.
FLOW_SKIP_SQL_ANALYZE	O modo de execução para instruções SQL. Valores válidos: `true`: Executa várias instruções SQL ao mesmo tempo. `false` (Padrão): Executa uma instrução SQL por vez. Nota Este parâmetro é suportado apenas para execuções de teste no ambiente de desenvolvimento de dados.
Outros	Você também pode adicionar parâmetros personalizados de job MR na seção de configuração avançada. Ao confirmar o código, o DataWorks adiciona automaticamente os novos parâmetros ao comando usando a instrução `-D key=value`.

Cluster Hadoop: EMR no ECS

Parâmetro	Descrição
queue	A fila onde os jobs são enviados. O valor padrão é default. Para mais informações sobre o EMR YARN, consulte Basic queue configurations.
priority	A prioridade do job. O valor padrão é 1.
USE_GATEWAY	Especifica se os jobs deste nó devem ser enviados por meio de um cluster de gateway. Valores válidos: `true`: Envia jobs por meio de um cluster de gateway. `false` (Padrão): Não envie jobs por meio de um cluster de gateway. Os jobs são enviados para o nó mestre por padrão. Nota Se você definir este parâmetro como `true`, mas o cluster do nó não estiver associado a um cluster de gateway, o envio do job EMR falhará.

Execute a tarefa

Em Run Configuration, em Compute Resource, configure Compute Resource e Resource Group.
Nota
- especifique as CUs for Scheduling de acordo com os requisitos da sua tarefa. O valor padrão é 0.25.
- Para acessar uma source de dados pela internet pública ou em uma Virtual Private cloud (VPC), use um grupo de recursos para agendamento que possa se conectar à source de dados. Para mais informações, consulte Network connectivity solutions.
Na caixa de diálogo de parâmetros na barra de ferramentas, selecione a source de dados criada e clique em Run.

Se precisar executar a tarefa do nó periodicamente, configure suas propriedades de agendamento. Para mais informações, consulte Configure scheduling properties for a node.
Após configurar o nó, faça o deploy dele. Para mais informações, consulte Deploy nodes.
Após o deploy da tarefa, visualize seu status no Operation Center. Para mais informações, consulte Introduction to Operation Center.

Visualize os resultados

acesse o console do OSS. visualize o arquivo de saída no diretório de destino dentro do seu bucket. Neste exemplo, o caminho é emr/datas/wordcount02/outputs.
Leia os resultados estatísticos no DataWorks.
1. Crie um nó EMR Hive. Para mais informações, consulte Create a node for a scheduled workflow.
2. No nó EMR Hive, crie uma tabela externa do Hive mapeada para os dados no OSS e, em seguida, consulte os dados da tabela. O código de exemplo é:
```
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;
```
  A figura a seguir mostra o resultado.