DataWorks:Criar um nó EMR MR

Preparar os dados de amostra

Crie um arquivo de amostra chamado input01.txt com o seguinte conteúdo:

hadoop emr hadoop dw
hive hadoop
dw emr

Carregar o arquivo de dados de amostra

1. Faça login no console do OSS. No painel de navegação à esquerda, clique em Buckets.

2. Clique no nome do bucket desejado para abrir a página File Management.

   Este tópico usa como exemplo um bucket chamado onaliyun-bucket-2.

3. Clique em New Directory para criar diretórios para os dados de amostra e os recursos JAR.

   • Defina Directory Name como emr/datas/wordcount02/inputs para criar o diretório dos dados de amostra.

   • Defina Directory Name como emr/jars para criar o diretório dos recursos JAR.

4. Carregue o arquivo de dados de amostra no diretório de dados.

   • 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 clique em Upload File.

Criar um pacote JAR MapReduce

1. Abra seu projeto no IntelliJ IDEA e adicione as seguintes dependências POM:

           <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. Para ler e gravar arquivos do OSS no MapReduce, configure os parâmetros abaixo.

   Importante

   Aviso de risco: A AccessKey da sua conta Alibaba Cloud tem acesso total a todas as APIs. Recomendamos o uso de um usuário RAM para acesso à API e operações diárias de O&M. Não armazene seu AccessKey ID e AccessKey Secret no código do projeto ou em locais onde possam ficar expostos. Uma AccessKey vazada pode comprometer todos os recursos da sua conta. O exemplo de código a seguir serve apenas como referência. 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 do OSS. O bucket do OSS deve estar na mesma região do cluster EMR, e essa região determina o endpoint. Para mais informações, consulte Regions and endpoints.

   O código Java a seguir modifica o exemplo WordCount do site oficial do Hadoop ao adicionar o AccessKey ID e o AccessKey secret para conceder ao job permissão de acesso ao arquivo no OSS.

   Código de exemplo

   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. Empacote o código em um arquivo JAR.

   Após editar e salvar o código Java, empacote-o em um arquivo JAR. Neste exemplo, o pacote JAR recebe o nome onaliyun_mr_wordcount-1.0-SNAPSHOT.jar.