Nodo EMR MR - DataWorks - Centro de documentación de Alibaba Cloud

En DataWorks, puede crear un nodo de MapReduce (MR) de E-MapReduce (EMR) para dividir conjuntos de datos grandes en tareas map paralelas. Esto mejora significativamente la eficiencia del procesamiento de datos. Este tema proporciona un ejemplo para desarrollar y configurar un trabajo de EMR MR que lee un archivo de texto desde Object Storage Service (OSS) y cuenta las palabras que contiene.

Requisitos previos

Cree un clúster de EMR de Alibaba Cloud y regístrelo en DataWorks. Para obtener más información, consulte New Data Studio: Attach an EMR compute resource.
(Opcional, para usuarios de RAM) El usuario de Resource Access Management (RAM) que se usa para el desarrollo de tareas debe agregarse al espacio de trabajo y asignársele el rol Development o Workspace Administrator (este rol incluye permisos amplios y debe otorgarse con precaución). Para obtener más información, consulte Add workspace members.

Si utiliza una cuenta raíz, omita este paso.
Para seguir el ejemplo de este tema, cree un bucket en Object Storage Service (OSS). Para obtener más información, consulte Create buckets.

Limitaciones

Este tipo de nodo solo se puede ejecutar en un serverless resource group (recomendado) o en un resource group exclusivo para programación.
Para gestionar los metadatos de un clúster DataLake o personalizado en DataWorks, primero debe configurar EMR-HOOK en el clúster. Para obtener más información, consulte Configure Hive EMR-HOOK.

Nota
Si no se configura EMR-HOOK en el clúster, DataWorks no podrá mostrar metadatos en tiempo real, generar registros de auditoría, mostrar el linaje de datos ni ejecutar tareas de gobernanza de datos relacionadas con EMR.

Preparar los datos iniciales y un paquete JAR

Preparar los datos iniciales

Cree un archivo de ejemplo llamado input01.txt con a continuación contenido.

hadoop emr hadoop dw
hive hadoop
dw emr

Cargar el archivo de datos iniciales

Inicie sesión en la consola de OSS. En el panel de navegación izquierdo, haga clic en Buckets.
Haga clic en el nombre del bucket de destino para abrir la página File Management.

Este ejemplo utiliza un bucket llamado onaliyun-bucket-2.
Haga clic en Create Directory para crear los directorios de los datos iniciales y del recurso JAR.
- Establezca Directory Name como emr/datas/wordcount02/inputs para crear el directorio de los datos iniciales.
- Establezca Directory Name como emr/jars para crear el directorio del recurso JAR.
Cargue el archivo de datos iniciales en su directorio.
- Vaya a la ruta /emr/datas/wordcount02/inputs y haga clic en Upload File.
- En el área Files to Upload, haga clic en Select Files, agregue el archivo input01.txt al bucket y luego haga clic en Upload File.

Compilar un trabajo de MapReduce y un paquete JAR

Abra su proyecto en IntelliJ IDEA y agregue a continuación dependencias a su archivo 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 leer y escribir en un archivo de OSS mediante MapReduce, debe configurar a continuación parámetros.

Importante

Advertencia de riesgo: El par de claves accesskey de su cuenta de Alibaba Cloud otorga acceso total a todas las operaciones de API. Exponer su AccessKey ID y su AccessKey Secret compromete la seguridad de todos los recursos de su cuenta. Se recomienda encarecidamente usar un usuario de RAM para las llamadas a API y las operaciones diarias. No codifique de forma rígida su AccessKey ID ni su AccessKey Secret en el código de su proyecto ni en ninguna otra ubicación de acceso público. A continuación código es solo para fines de demostración. Mantenga segura su información de accesskey.

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

A continuación tabla describe los parámetros.

${accessKeyId}: El AccessKey ID de su cuenta de Alibaba Cloud.
${accessKeySecret}: El AccessKey Secret de su cuenta de Alibaba Cloud.
${endpoint}: El endpoint público de OSS. El endpoint depende de la región donde reside su clúster de EMR. El bucket de OSS y el clúster deben estar en la misma región. Para obtener más información, consulte Regions and endpoints.

A continuación código Java es una versión modificada del ejemplo oficial WordCount de Hadoop. Incluye las configuraciones del AccessKey ID y del AccessKey Secret para otorgar al trabajo el permiso necesario para acceder a los archivos de OSS.

Código de ejemplo

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

Después de editar el código Java, compílelo en un archivo JAR. Este ejemplo genera un archivo JAR llamado onaliyun_mr_wordcount-1.0-SNAPSHOT.jar.

Procedimiento

En la pestaña de configuración del nodo EMR MR, desarrolle la tarea de la siguiente manera:

Desarrollar la tarea de EMR MR

Elija uno de a continuación métodos según sus necesidades:

Método 1: Cargar y hacer referencia al archivo JAR

También puede cargar un recurso desde su computadora local a DataStudio y luego hacer referencia a él en un nodo. Si un recurso es demasiado grande para cargarse mediante la consola de DataWorks, puede almacenarlo en HDFS y referenciarlo en su código.

Cree un recurso JAR.
1. Para obtener más información, consulte Manage resources. Almacene el paquete JAR del paso Prepare initial data and a JAR package en el directorio emr/jars. Haga clic en Click Upload.
2. Configure los parámetros Storage Path, Data Sources y Resource Group.
3. Haga clic en Save.
Haga referencia al recurso JAR.
1. Abra el nodo EMR MR para ir a su pestaña de configuración.
2. En el panel Resource Management de la izquierda, busque el recurso al que desea hacer referencia. En este ejemplo, el recurso es onaliyun_mr_wordcount-1.0-SNAPSHOT.jar. Haga clic con el botón derecho en el recurso y seleccione Insert Resource Path.
3. Una vez que hace referencia al recurso, aparece una instrucción de referencia en la pestaña de configuración del nodo EMR MR. Esto indica que el recurso se referenció correctamente. Ejecute a continuación comando. Reemplace el paquete de recursos, el nombre del bucket y la ruta con su información real.
```
##@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
  El editor de código de los nodos EMR MR no admite comentarios.

Método 2: Hacer referencia a un recurso de OSS

Puede usar el método OSS REF para hacer referencia directa a un recurso de OSS. Cuando se ejecuta el nodo, DataWorks descarga automáticamente el recurso de OSS referenciado al entorno local. Este método se usa frecuentemente en escenarios donde una tarea de EMR depende de un archivo JAR o de un script.

Cargue el recurso JAR.
1. Después de desarrollar el código, inicie sesión en la consola de OSS. En el panel de navegación izquierdo, haga clic en Buckets.
2. Haga clic en el nombre del bucket de destino para abrir la página File Management.
  
  Este ejemplo utiliza un bucket llamado onaliyun-bucket-2.
3. Cargue el recurso JAR en su directorio.
  
  Vaya al directorio emr/jars. Haga clic en Upload File. En el área Files to Upload, haga clic en Select Files, agregue el archivo onaliyun_mr_wordcount-1.0-SNAPSHOT.jar y luego haga clic en Upload File.

Haga referencia al recurso JAR.

En la pestaña de configuración del nodo EMR MR, escriba el código para hacer referencia al 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

El comando tiene a continuación formato: hadoop jar <Path of the JAR to run> <Full name of the main class> <Input file directory> <Output directory>.

A continuación tabla describe los parámetros de la ruta del archivo JAR.

Parámetro

Descripción

Ruta del JAR a ejecutar

El formato es ossref://{endpoint}/{bucket}/{object}

Endpoint: El endpoint público de OSS. Si este parámetro se deja vacío, solo podrá hacer referencia a recursos de un bucket que esté en la misma región que el clúster de EMR.
Bucket: Un contenedor que OSS utiliza para almacenar objetos. Cada Bucket tiene un nombre único. Puede iniciar sesión en la consola de OSS para consultar todos los Buckets de su cuenta.
object: Un objeto específico, que puede ser un archivo o una ruta, almacenado en un bucket.

(Opcional) Configurar parámetros avanzados

En el panel derecho, haga clic en la pestaña Scheduling Settings. Puede configurar a continuación parámetros en la sección EMR Node Parameters > DataWorks parameters.

Nota

Los parámetros avanzados disponibles varían según el tipo de clúster de EMR, como se muestra en a continuación tablas.
Configure propiedades adicionales de Spark de código abierto en la pestaña Scheduling Settings, dentro de la sección EMR Node Parameters > Spark parameter.

Clúster DataLake y clúster personalizado: EMR on ECS

Parámetro	Descripción
queue	La cola donde se envían los trabajos. El valor predeterminado es default. Para obtener más información sobre EMR YARN, consulte Basic queue configurations.
priority	La prioridad del trabajo. El valor predeterminado es 1.
FLOW_SKIP_SQL_ANALYZE	El modo de ejecución de las instrucciones SQL. Valores válidos: `true`: Ejecuta varias instrucciones SQL a la vez. `false` (predeterminado): Ejecuta una instrucción SQL a la vez. Nota Este parámetro solo es compatible con ejecuciones de prueba en el entorno de desarrollo de datos.
Otros	También puede agregar parámetros personalizados del trabajo de MR en la sección de configuración avanzada. Cuando envíe el código, DataWorks agregará automáticamente los nuevos parámetros al comando mediante la instrucción `-D key=value`.

Clúster de Hadoop: EMR on ECS

Parámetro	Descripción
queue	La cola donde se envían los trabajos. El valor predeterminado es default. Para obtener más información sobre EMR YARN, consulte Basic queue configurations.
priority	La prioridad del trabajo. El valor predeterminado es 1.
USE_GATEWAY	Especifica si los trabajos de este nodo se envían a través de un clúster de puerta de enlace. Valores válidos: `true`: Envía los trabajos a través de un clúster de puerta de enlace. `false` (predeterminado): No envía los trabajos a través de un clúster de puerta de enlace. De forma predeterminada, los trabajos se envían al nodo maestro. Nota Si establece este parámetro en `true`, pero el clúster del nodo no está asociado con un clúster de puerta de enlace, el envío del trabajo de EMR fallará.

Ejecutar la tarea

En Run Configuration, dentro de Compute Resource, configure Compute Resource y Resource Group.
Nota
- También puede especificar CUs for Scheduling según los requisitos de su tarea. El valor predeterminado es 0.25.
- Para acceder a un origen de datos a través de internet público o de una Virtual Private Cloud (VPC), debe usar un resource group para programación que pueda conectarse a dicho origen de datos. Para obtener más información, consulte Network connectivity solutions.
En el cuadro de diálogo de parámetros de la barra de herramientas, seleccione el origen de datos que creó y haga clic en Run.

Si necesita ejecutar la tarea del nodo periódicamente, configure sus propiedades de programación. Para obtener más información, consulte Configure scheduling properties for a node.
Después de configurar el nodo, debe desplegarlo. Para obtener más información, consulte Deploy nodes.
Una vez desplegada la tarea, puede consultar su estado en Operation Center. Para obtener más información, consulte Introduction to Operation Center.

Consultar los resultados

Inicie sesión en la consola de OSS. Puede consultar el archivo de salida en el directorio de destino dentro de su bucket. En este ejemplo, la ruta es emr/datas/wordcount02/outputs.
Lea los resultados estadísticos en DataWorks.
1. Cree un nodo EMR Hive. Para obtener más información, consulte Create a node for a scheduled workflow.
2. En el nodo EMR Hive, cree una tabla externa de Hive asignada a los datos en OSS y luego consulte los datos de la tabla. A continuación se muestra un código de ejemplo:
```
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 continuación figura muestra el resultado.