Create an EMR MR node - DataWorks - Centre de documentation Alibaba Cloud

Un nœud E-MapReduce (EMR) MR divise un grand ensemble de données en plusieurs tâches map parallèles pour accélérer les calculs à grande échelle. Cette rubrique utilise un exemple WordCount qui lit du texte depuis Object Storage Service (OSS) et compte les mots afin de vous guider tout au long du flux de développement d'un nœud EMR MR.

Prérequis

Un cluster EMR doit être enregistré dans DataWorks. Pour plus d'informations, consultez la rubrique DataStudio (ancienne version) : Associer une ressource de calcul EMR.
(Facultatif, pour les utilisateurs RAM) L'utilisateur RAM chargé du développement des tâches doit être membre de l'espace de travail et disposer du rôle Development ou Workspace Administrator. Le rôle Workspace Administrator dispose d'autorisations étendues ; attribuez-le avec prudence. Pour plus d'informations, consultez la rubrique Ajouter des membres à un espace de travail.
Un groupe de ressources serverless doit être acheté et configuré, y compris l'association à l'espace de travail et les paramètres réseau. Pour plus d'informations, consultez la rubrique Utiliser un groupe de ressources serverless.
Un workflow doit être créé dans Data Development (DataStudio). Pour plus d'informations, consultez la rubrique Créer un workflow.
Si la tâche de votre nœud EMR MR doit référencer du code open source, vous devez d'abord télécharger le code en tant que ressource vers un nœud de ressource EMR JAR. Pour plus d'informations, consultez la rubrique Créer et utiliser des ressources EMR.
Si la tâche de votre nœud EMR MR doit référencer une fonction définie par l'utilisateur (UDF), vous devez d'abord télécharger l'UDF en tant que ressource vers un nœud de ressource EMR JAR, puis créer et enregistrer la fonction. Pour plus d'informations, consultez la rubrique Créer une fonction EMR.
Si vous exécutez l'exemple de développement de tâche présenté dans cette rubrique, vous devez d'abord créer un bucket OSS. Pour plus d'informations, consultez la rubrique Créer un bucket dans la console.

Limites

Ce type de tâche ne peut s'exécuter que sur un groupe de ressources serverless (recommandé) ou un groupe de ressources exclusif pour la planification.
Pour gérer les métadonnées d'un cluster DataLake ou personnalisé dans DataWorks, vous devez d'abord configurer EMR-HOOK sur le cluster. Sans EMR-HOOK, DataWorks ne peut pas afficher les métadonnées en temps réel, générer des journaux d'audit, montrer la lignée des données ni effectuer des tâches de gouvernance liées à EMR. Pour savoir comment configurer EMR-HOOK, consultez la rubrique Configurer EMR-HOOK pour Hive.

Préparer les données d'exemple et le package JAR

Préparer les données d'exemple

Créez un fichier d'exemple nommé input01.txt contenant le texte suivant.

hadoop emr hadoop dw
hive hadoop
dw emr

Télécharger le fichier de données d'exemple

Connectez-vous à la console OSS. Dans le volet de navigation de gauche, cliquez sur Buckets.
Cliquez sur le nom du bucket cible pour ouvrir la page File Management.

Cette rubrique utilise un bucket nommé onaliyun-bucket-2 à titre d'exemple.
Cliquez sur New Directory pour créer des répertoires destinés aux données d'exemple et aux ressources JAR.
- Définissez Directory Name sur emr/datas/wordcount02/inputs afin de créer un répertoire pour les données d'exemple.
- Définissez Directory Name sur emr/jars afin de créer un répertoire pour les ressources JAR.
Téléchargez le fichier de données d'exemple dans le répertoire de données.
- Accédez au chemin /emr/datas/wordcount02/inputs et cliquez sur Upload File.
- Dans la zone Files to Upload, cliquez sur Select Files, ajoutez le fichier input01.txt au bucket, puis cliquez sur Upload File.

Générer un package MapReduce JAR

Ouvrez votre projet IntelliJ IDEA et ajoutez les dépendances POM suivantes.

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

Pour lire et écrire des fichiers OSS dans MapReduce, vous devez configurer les paramètres suivants.

Important

Avertissement de sécurité : La clé AccessKey de votre compte Alibaba Cloud dispose d'un accès complet à toutes les API. Nous recommandons d'utiliser un utilisateur RAM pour l'accès aux API et les opérations quotidiennes. Ne stockez pas votre AccessKeyID et votre AccessKeySecret dans le code de votre projet ni dans tout emplacement susceptible d'être exposé. Une clé AccessKey divulguée peut compromettre toutes les ressources associées à votre compte. L'exemple de code ci-dessous est fourni à titre indicatif uniquement. Protégez vos informations AccessKey.

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

Le tableau suivant décrit les paramètres.

${accessKeyId} : ID de la clé AccessKey de votre compte Alibaba Cloud.
${accessKeySecret} : Secret de la clé AccessKey de votre compte Alibaba Cloud.
${endpoint} : Endpoint OSS. Le bucket OSS doit se trouver dans la même région que le cluster EMR, et cette région détermine l'endpoint. Pour plus d'informations, consultez la rubrique Régions et endpoints.

Le code Java suivant modifie l'exemple WordCount du site officiel Hadoop en ajoutant l'AccessKeyID et l'AccessKeySecret afin d'autoriser la tâche à accéder au fichier OSS.

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

Empaquetez le code dans un fichier JAR.

Après avoir modifié et enregistré le code Java, empaquetez-le dans un fichier JAR. Dans cet exemple, le package JAR est nommé onaliyun_mr_wordcount-1.0-SNAPSHOT.jar.

Étape 1 : Créer un nœud EMR MR

Connectez-vous à la console DataWorks. Dans la région cible, cliquez sur Data Development and O&M > Data Development dans le volet de navigation de gauche. Sélectionnez un espace de travail dans la liste déroulante et cliquez sur Go to Data Development.
Créez un nœud EMR MR.
1. Cliquez avec le bouton droit sur le workflow cible et choisissez Create Node > EMR > EMR MR.
  
  Remarque
  Vous pouvez également placer le pointeur sur l'icône Create et choisir Create Node > EMR > EMR MR.
2. Dans la boîte de dialogue Create Node, saisissez un Name, puis sélectionnez une Engine Instance, un Node Type et un Path. Cliquez sur Confirm pour ouvrir l'éditeur de nœud EMR MR.
  
  Remarque
  Les noms de nœuds peuvent contenir des lettres, des chiffres, des traits de soulignement (_) et des points (.).

Étape 2 : Développer la tâche EMR MR

Dans l'éditeur de nœud EMR MR, double-cliquez sur le nœud pour ouvrir la page de développement de la tâche. Choisissez l'une des options suivantes selon votre scénario :

(Recommandé) Téléchargez d'abord les ressources depuis votre machine locale vers DataStudio, puis référencez-les. Pour plus d'informations, consultez la section Option 1 : Télécharger et référencer une ressource.
Référencez les ressources OSS à l'aide de la méthode OSS REF. Pour plus d'informations, consultez la section Option 2 : Utiliser OSS REF.

Option 1 : Télécharger et référencer une ressource

Téléchargez les ressources depuis votre machine locale vers DataStudio et référencez-les dans votre nœud. Si une ressource est trop volumineuse pour être téléchargée via la console, stockez-la dans HDFS et référencez-la directement depuis votre code.

Créez une ressource EMR JAR.

Pour plus d'informations, consultez la rubrique Créer et utiliser une ressource EMR. Dans cet exemple, le package JAR généré dans la section Préparer les données d'exemple et le package JAR est stocké dans le répertoire emr/jars. La première fois que vous utilisez cette fonctionnalité, cliquez sur Authorize, puis sur Click Upload pour télécharger la ressource JAR. Dans l'arborescence du workflow située à gauche, cliquez avec le bouton droit sur Resource pour ouvrir la boîte de dialogue Create Resource. Définissez Engine Type sur EMR et Resource Type sur EMR JAR. Pour Storage Path, sélectionnez One-click authorization for OSS. Téléchargez le fichier, définissez son Name sur onaliyun_mr_wordcount-1.0-SNAPSHOT.jar, puis cliquez sur Create.
Référencez la ressource EMR JAR.
1. Ouvrez le nœud EMR MR que vous avez créé et accédez à l'éditeur de code.
2. Dans le dossier EMR > Resources, recherchez la ressource à référencer, par exemple onaliyun_mr_wordcount-1.0-SNAPSHOT.jar dans cet exemple. Cliquez avec le bouton droit sur la ressource et choisissez Insert Resource Path.
3. Une fois la référence sélectionnée, un message de réussite s'affiche sur la page d'édition du code du nœud EMR MR. Cela indique que la ressource de code est correctement référencée. Exécutez ensuite la commande suivante. Les packages de ressources, les noms de bucket et les informations de chemin dans la commande ci-dessous sont donnés à titre d'exemple. Vous devez les remplacer par vos propres informations.
```
##@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
```
  Remarque
  L'éditeur de code des nœuds EMR MR ne prend pas en charge les instructions de commentaire.

Option 2 : Utiliser OSS REF

La méthode OSS REF référence directement une ressource depuis OSS. Lors de l'exécution du nœud, DataWorks charge automatiquement la ressource OSS spécifiée pour la tâche. Cette méthode convient lorsqu'une tâche EMR dépend d'un JAR ou d'un script.

Téléchargez la ressource JAR.
1. Après avoir développé le code, connectez-vous à la console OSS. Dans le volet de navigation de gauche correspondant à votre région, cliquez sur Buckets.
2. Cliquez sur le nom du bucket cible pour ouvrir la page File Management.
  
  Cette rubrique utilise un bucket nommé onaliyun-bucket-2 à titre d'exemple.
3. Téléchargez la ressource JAR dans son répertoire de stockage.
  
  Accédez au répertoire emr/jars et cliquez sur Upload File. Dans la zone Files to Upload, cliquez sur Select Files, ajoutez le fichier onaliyun_mr_wordcount-1.0-SNAPSHOT.jar au bucket, puis cliquez sur Upload File.

Référencez la ressource JAR.

Modifiez le code pour référencer la ressource JAR sur la page de configuration du nœud EMR MR.

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

Remarque

Le format de la commande est le suivant : hadoop jar <path_of_the_JAR_file_to_run> <fully_qualified_name_of_the_main_class> <input_directory> <output_directory>.

Le tableau suivant décrit le paramètre relatif au chemin du fichier JAR.

Paramètre

Description

Chemin du fichier JAR à exécuter

Le format est ossref://{endpoint}/{bucket}/{object}.

Endpoint : Endpoint OSS. Ce paramètre est facultatif. S'il est omis, le système utilise par défaut la région de votre cluster EMR, et vous ne pouvez accéder qu'aux ressources OSS situées dans cette même région.
Bucket : Conteneur permettant de stocker des objets dans OSS. Chaque bucket possède un nom unique. Vous pouvez vous connecter à la console OSS pour afficher tous les buckets associés à votre compte.
object : Objet spécifique, tel qu'un nom de fichier ou un chemin, stocké dans un bucket.

(Facultatif) Configurer les paramètres avancés

Configurez les propriétés spécifiques au nœud dans l'onglet Advanced Settings. Pour plus d'informations sur ces propriétés, consultez la documentation Spark Configuration. Les paramètres avancés disponibles varient selon le type de cluster EMR, comme indiqué dans les tableaux suivants.

Cluster DataLake et personnalisé

Paramètre avancé

Description

queue

File d'attente de planification à laquelle la tâche est soumise. La valeur par défaut est default. Pour plus d'informations sur EMR YARN, consultez la rubrique Configurations de file d'attente de base.

priority

Priorité de la tâche. La valeur par défaut est 1.

Autre

Vous pouvez ajouter des paramètres personnalisés pour les tâches MR dans l'onglet Advanced Settings. Lors de la validation du code, DataWorks ajoute automatiquement ces paramètres à la commande au format -D key=value.

Cluster Hadoop

Paramètre avancé	Description
queue	File d'attente de planification à laquelle la tâche est soumise. La valeur par défaut est `default`. Pour plus d'informations sur EMR YARN, consultez la rubrique Configurations de file d'attente de base.
priority	Priorité de la tâche. La valeur par défaut est 1.
USE_GATEWAY	Indique si la tâche doit être soumise via un cluster gateway. `true` : soumet la tâche via un cluster gateway. `false` (par défaut) : ne soumet pas la tâche via un cluster gateway. La tâche est soumise au nœud master par défaut. Remarque Si le cluster hébergeant le nœud n'est pas associé à un cluster gateway et que vous définissez manuellement ce paramètre sur `true`, la soumission de la tâche EMR échoue.

Exécuter la tâche

Dans la barre d'outils, cliquez sur l'icône . Dans la boîte de dialogue Parameter, sélectionnez le groupe de ressources de planification et cliquez sur Running.
Remarque
- Pour accéder à une ressource de calcul via un réseau public ou un VPC, vous devez utiliser un groupe de ressources de planification ayant réussi le test de connectivité avec la ressource de calcul. Pour plus d'informations, consultez la rubrique Solutions de connectivité réseau.
- Si vous devez modifier le groupe de ressources pour les tâches suivantes, vous pouvez cliquer sur l'icône Run with Parameters et sélectionner le groupe de ressources souhaité.
Cliquez sur l'icône pour enregistrer le code.
(Facultatif) Effectuez un test de fumée.

Pour effectuer un test de fumée dans l'environnement de développement, vous pouvez le faire avant ou après avoir validé le nœud. Pour plus d'informations, consultez la rubrique Effectuer un test de fumée.

Étape 3 : Configurer les propriétés de planification

Si vous souhaitez que le système exécute périodiquement une tâche sur le nœud, vous pouvez cliquer sur Properties dans le volet de navigation de droite de l'onglet de configuration du nœud afin de configurer les propriétés de planification des tâches selon vos besoins métier. Pour plus d'informations, consultez la rubrique Vue d'ensemble.

Remarque

Vous devez configurer les paramètres Rerun et Parent Nodes dans l'onglet Properties avant de valider la tâche.

Étape 4 : Déployer la tâche

Une fois la tâche d'un nœud configurée, vous devez la valider et la déployer. Après validation et déploiement, le système exécute la tâche régulièrement selon les configurations de planification.

Cliquez sur l'icône dans la barre d'outils supérieure pour enregistrer la tâche.
Cliquez sur l'icône dans la barre d'outils supérieure pour valider la tâche.

Dans la boîte de dialogue Submit, configurez le paramètre Change description. Ensuite, décidez si vous souhaitez examiner le code de la tâche après sa validation, en fonction de vos besoins métier.
Remarque
- Vous devez configurer les paramètres Rerun et Parent Nodes dans l'onglet Properties avant de valider la tâche.
- Vous pouvez utiliser la fonctionnalité de révision de code pour garantir la qualité du code des tâches et prévenir les erreurs d'exécution dues à un code invalide. Si vous activez la fonctionnalité de révision de code, le code de la tâche validé ne pourra être déployé qu'après avoir passé avec succès la révision. Pour plus d'informations, consultez la rubrique Révision de code.

Si vous utilisez un espace de travail en mode standard, vous devez déployer la tâche dans l'environnement de production après l'avoir validée. Pour déployer une tâche sur un nœud, cliquez sur Deploy dans le coin supérieur droit de l'onglet de configuration du nœud. Pour plus d'informations, consultez la rubrique Déployer un nœud.

Opérations supplémentaires

Après validation et déploiement de la tâche, celle-ci s'exécute périodiquement selon les configurations de planification. Vous pouvez cliquer sur Operation Center dans le coin supérieur droit de l'onglet de configuration du nœud correspondant pour accéder à Operation Center et consulter l'état de planification de la tâche. Pour plus d'informations, consultez la rubrique Gérer les tâches planifiées.

Consulter les résultats

Connectez-vous à la console OSS. Consultez la sortie dans le répertoire spécifié de votre bucket. Chemin d'exemple : emr/datas/wordcount02/outputs. Une fois la tâche wordcount02 terminée, un fichier indicateur _SUCCESS et plusieurs fichiers de fragments de résultat part-r-* sont générés dans le répertoire /emr/datas/wordcount02/outputs/ d'OSS, confirmant que la tâche a produit une sortie avec succès.
Lisez les statistiques dans DataWorks.
1. Créez un nœud EMR Hive. Pour plus d'informations, consultez la rubrique Créer un nœud EMR Hive.
2. Dans le nœud EMR Hive, créez une table externe Hive basée sur les données présentes dans OSS. Voici un exemple de code :
```
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;
```
  Après exécution de la requête, l'onglet Results affiche les statistiques de fréquence des mots : dw (2), hadoop (3), emr (2) et hive (1).