EMR : Instancier son cluster Hadoop simplement

Amazon Elastic MapReduce (EMR) est le service qui simplifie la création et la mise à disposition d’une stack Hadoop complète et ajustable. Grâce à ce service, il suffit de quelques opérations pour créer un cluster Hadoop dans le Cloud embarquant les principaux outils de l’écosystème tels que YARN, Spark, Zeppelin, Hive, Presto, HBase et HDFS. Cet article propose de faire un tour d’horizon de ce service et de quelques moyens d’en profiter. Il s’inscrit dans une série d’articles autour de l’instanciation de clusters dans le Cloud.

Nous présenterons les raisons qui peuvent pousser à utiliser EMR, puis nous parlerons brièvement des différents services Amazon impliqués. Ensuite, après avoir déterminé l’architecture cible, nous passerons en revue les questions qu’il est indispensable de se poser, puis nous créerons notre cluster.

Pourquoi utiliser EMR ?

Il est intéressant d’utiliser EMR pour limiter le coût de l’infrastructure nécessaire à l’exécution des services Hadoop. Dans la plupart des cas, l’infrastructure englobant l’écosystème Hadoop n’est utile que durant la phase de processing de la donnée. Une fois le traitement terminé, l’infrastructure peut être détruite.

Voici la chaîne d’opérations à réaliser dans le cadre de traitements de type batch.

1. Stocker les données dans le Cloud.
2. Instancier un cluster Hadoop.
3. Traiter les données au sein du cluster.
4. Produire de nouvelles données ayant de la valeur.
5. Stocker les données utiles résultantes dans le Cloud.
6. Détruire le cluster.
7. Exploiter les résultats.

La solution EMR est flexible en terme d’infrastructure, de distribution Hadoop, de produits logiciels inclus et de durée de vie du cluster. De plus, le monitoring du cluster est inclus avec CloudWatch. Pour finir, EMR offre une interaction avec l’ensemble des autres services AWS.

L’utilisation d’EMR n’est pas restreinte à des chaînes d’utilisation de type batch. EMR est capable de consommer un nombre important de ressources variées, ce qui ouvre la porte à de nombreux usages.

EMR fait intervenir un ensemble de services AWS qu’il est bon d’introduire.

Amazon Virtual Private Cloud (VPC)

Tous les noeuds EMR s’exécutent dans un VPC au sein d’un sous réseau d’une zone de disponibilité (AZ). Pour rappel, en fonction d’une zone géographique où le service AWS est disponible comme par exemple en Europe (Dublin et bientôt Paris), il existe deux à trois zones de disponibilité (AZ). Pour y voir plus clair sur les VPC et avoir une vision complète, rien de tel que l’article de mon collègue Jeremy Pinsolle.

Identity and Access Management (IAM)

C’est le service qui permet la gestion des identités et des accès. Ce service pourrait faire l’objet d’un article à lui tout seul ! Pour rester sur notre sujet EMR, nous allons retenir que pour instancier le cluster EMR via API REST, il est nécessaire d’activer une clé d’API dans IAM pour notre compte utilisateur AWS.

IAM permet d’attacher des stratégies de sécurité (policies) relatives à l’utilisation des services. Chaque stratégie est identifiable via un nom unique Amazon Resource Name (ARN) et peut être associée à des utilisateurs ou à des rôles. Ce système d’ARN est utilisé partout sur AWS, où tout est considéré comme une ressource.

Voici un exemple de stratégie permettant à l’utilisateur de changer son propre mot de passe dans IAM :

[java]arn:aws:iam::aws:policy/IAMUserChangePassword[/java]

Au sein d’ IAM, des droits sont associés à cette stratégie. Dans le cas présent, si j’associe cette stratégie à mon compte utilisateur et que j’ajoute des droits de lecture/écriture, je pourrai écrire ou lire sur le service IAM pour changer mon mot de passe.

Pour EMR, nous allons devoir créer deux rôles dans IAM.

Un premier rôle pour le service EMR : Celui-ci permettra à EMR de créer un subnet, et d’instancier des machines avec EC2.

[java]arn:aws:iam::XXXXX:role/EMR_DefaultRole[/java]

Un second rôle pour les instances qui composent le cluster EMR pour qu’elles puissent par exemple récupérer des données sur S3 et consommer des services tiers.

[java]arn:aws:iam::XXXXX:role/EMR_EC2_DefaultRole[/java]

EC2

Ce sont les instances de machines qui composent le cluster Hadoop. Il convient de choisir les types d’instances EC2 adéquates au regard du nombre de composants déployés et des performances attendues. Amazon fournit la liste des instances compatibles avec EMR. A titre d’exemple, une m3.xlarge avec ses 4vCPU, 15Go de RAM et 2*40 Go de disque SSD suffit à faire tourner une typologie de cluster à base de Spark, Hadoop, Tez, Hive avec un niveau de performance acceptable. 

S3

Le service de stockage historique est utilisé pour stocker les logs d’EMR, ainsi que les données froides en entrée et en sortie du cluster. Il est enfin utilisé pour stocker les Notebooks Zeppelin si on utilise ce composant. Ajouté à ces usages, on peut l’utiliser pour stocker les données d’HBase  sans HDFS.

Amazon Machine Image (AMI)

Il s’agit des images de machines utilisées pour peupler notre cluster. L’utilisation de l’AMI AWS permet de ne pas ajouter de coûts supplémentaire de licence.

Dynamo DB

Le composant EMR FS, dont l’utilisation permet de s’assurer de la cohérence et de la localisation des données sur S3, utilise Dynamo DB pour stocker en interne ses métadonnées. Avec lui, EMR consomme facilement les données de S3.

Relational Database Service (RDS)

RDS est fortement recommandé pour stocker les données non reconstructibles telles que :

• les informations du méta store Hive
• les Workflows Oozie
• les métadonnées de Hue

Grâce à cela, même si j’utilise un cluster éphémère mes métadonnées ne seront pas perdues.

Elastic Block Store (EBS)

Les instances EC2 sont livrées avec des capacités disque différentes. Par exemple, pour des instances de type m3.xlarge, elles reposent sur deux disques SSD de 40 Go. Dans un contexte EMR, une partie de cet espace disque est dédié au système d’exploitation et le reste est dédié à HDFS. La capacité d’HDFS est ajustable via l’ajout d’EBS supplémentaires aux instances.

Architecture cible 

L’architecture cible choisie est simple et n’aborde pas les aspects sécurité. Il conviendra d’ajouter ces composants en amont de l’architecture (bastion, Reverse proxy, Key Management Service, Security group…).

• Stockage : Les données sont déposées sur S3 avant l’instanciation. Notre cluster EMR log également son état sur S3.
• Data management : Toutes les tables créées par l’utilisateur dans Hive, par exemple, sont stockées dans RDS. Hive ou Presto utilisent ce que l’on nomme le schema on read, et par conséquent, peuvent poser un schéma sur de la donnée présente dans S3.
• Traitement : EMR consomme S3, et s’aide d’EMR FS pour s’assurer de la data locality en utilisant DynamoDB. Différents types de processing peuvent être réalisés via des outils tels que Hive, Spark, Zeppelin, Pig, Presto, HBase. Le cluster EMR expose un ensemble de WebUI.
• Stockage & exposition : Les résultats du traitement sont stockés dans S3. 

Préparation

Nous allons maintenant nous poser les bonnes questions pour préparer un cluster conforme à nos besoins.

De quelles données ai-je besoin ? Quel niveau de confidentialité ?

Avant d’instancier un cluster pour une utilisation en mode batch, il convient de copier les données utiles dans S3 et d’en permettre l’accès à notre futur cluster EMR pour qu’il les consomme. De nombreuses solutions existent pour amener la donnée sur S3. Vous en trouverez quelques exemples dans ce retour d’expérience d’un workshop Big Data. La manipulation de données implique le respect des différentes règles relatives à leur confidentialité et à leur protection. A titre d’exemple, les mécanismes de chiffrement des données S3 peuvent être activés. EMR FS supporte ces mécanismes à condition d’en utiliser qu’un seul simultanément.

Au niveau européen, la législation applicable « General Data Protection Regulation » (GDPR) pourra être respectée dès son entrée en vigueur en mai 2018, c’est du moins ce à quoi s’engage AWS.

De quelles type d’authentification utilisateur ai-je besoin ? De quelle manière manière dois-je gérer les autorisations ?

Les mécanismes habituels d’authentification utilisateur comme Kerberos ou encore LDAP peuvent être activés. Les « Role Based Access Control » (RBAC) avec l’outil Apache Ranger peuvent être implémentées ce qui permettra de gérer finement les accès utilisateur à vos données et de savoir qui y a accédé.

Quelle distribution choisir ?

Le service EMR embarque par défaut une distribution Hadoop composée des différents outils de l’écosystème Hadoop. La figure ci-dessous extraite de la documentation présente les différentes releases d’EMR avec les versions des outils correspondant :

Les outils qui composent une release d’EMR sont déclinés en version Open source pure ou en version spécifique pour EMR.

Un exemple illustrant ce point avec la bibliothèque « hadoop command line client » disponible en version 2.7.3-amzn-2 sur EMR. Dans ce cas, il s’agit d’une version spécifique pour EMR car sa signature est suffixée « amzn-2 ». Le choix de la distribution est motivé par les versions des composants que nous souhaitons utiliser, leurs disponibilité au sein d’une release EMR et la stabilité vérifiée de l’ensemble. Il convient de ne pas prendre la toute dernière release d’EMR mais l’une des plus récentes. Pour conclure ce choix, un cluster EMR peut être également instancié avec différentes versions de la distribution MapR. Le présent article ne couvre pas ce mode de fonctionnement.

De quels outils ai-je réellement besoin ?

À partir de la version EMR choisie, il faut sélectionner la liste des outils que nous allons utiliser. Si nous utilisons uniquement les outils nécessaires, notre cluster démarrera beaucoup plus rapidement. Par défaut, la console AWS propose quatre topologies de cluster. Via le client AWS en ligne de commande (aws-cli) ou au sein des options avancées, nous pouvons construire une topologie personnalisée.

La flexibilité d’EMR permet d’installer un composant à postériori avec une commande du type :

[java]aws emr install-applications –cluster-id j-2A6HXXXXXXL7J –apps Name=Hive[/java]

Ai-je besoin d’un cluster permanent ou éphémère ?

Nous pouvons lancer deux types de cluster EMR :

Les clusters « long running » sont permanents et ne s’éteignent pas sauf action spécifique de son propriétaire. Les clusters éphémères quant à eux réalisent une ou plusieurs étapes et s’éteignent une fois ces dernières terminées. Les étapes sont des mécanismes complets qui peuvent réaliser un ensemble d’actions comme par exemple un job Spark. On peut aussi ajouter des étapes sur les clusters « long running ». Historiquement, il existait une limitation à 256 étapes maximum sur un « long running » cluster. Cette règle est maintenant transformée en 256 étapes simultanées maximum sans limitations du nombre total d’étapes.

Le diagramme ci-dessous extrait de la documentation AWS présente les différents états d’un cluster EMR quelque soit son type.

On constatera que ce qui différencie le « long running » de l’éphémère sur ce diagramme est le fait que le long running reste dans l’état WAITING jusqu’à une action de terminaison du cluster. Un cluster éphémère quant à lui réalise ses étapes et passe ensuite au statut SHUTTING_DOWN puis COMPLETED.

Quelle est la volumétrie des données à traiter ? Quel type de traitement vais-je réaliser ?

La volumétrie de données à traiter est un facteur déterminant sur le sizing du cluster. Plus, il y a de données en entrée de mon traitement et plus je vais avoir besoin de ressources si je souhaite que l’ensemble de mon jeu de données soit traité en parallèle et rapidement.

Est-ce que mon traitement est CPU, IO ou Memory bound ? En fonction de ces critères, je vais choisir le type d’instance EC2 qui vont peupler mon cluster.

Un cluster EMR est composé de trois groupes d’instance(s) EC2 :

• Master : Noeuds maîtres ou noeuds de management. C’est sur ce type de noeud que des processus « Master » tels que le namenode de HDFS ou le ressource manager de YARN s’exécutent.
• Core : Ce sont les workers de notre cluster et ce sont eux qui exécutent le processus DataNode pour HDFS et Node Manager pour YARN.
• Task : Noeuds de pur calcul, ils peuvent exécuter des processus de type Node Manager pour YARN mais ne stockent pas les données d’HDFS.

La figure ci-dessous présente un exemple de topologie de cluster EMR incluant des noeuds core et master.

Quel est mon budget ?

Le budget est bien entendu primordial et ce d’autant plus lorsqu’on démarre une horde de machines EC2. Avant d’utiliser le service EMR, il est bon d’exploiter la calculatrice pour estimer le coût de facturation en se basant sur le coût horaire des machines. Il faut aussi tracer la facturation liée à l’instanciation du cluster en le « taggant ».

Afin de limiter les coûts, rien de tel que les instances ponctuelles, également nommées instances spot. On peut composer un cluster EMR uniquement à base d’instances ponctuelles. Pour ce faire, il faut se renseigner sur le cours historique de l’instance ciblée dans la zone de disponibilité cible. Si on propose un prix légèrement au dessus de l’historique du cours, on peut diviser par 5 le coût de l’instance à l’heure !

Voici une commande à adapter permettant de se renseigner sur le cours d’une instance de type « m3.2xlarge » depuis le 25/07/2017 à 7h08 dans l’AZ eu-west-1c

[java] aws ec2 describe-spot-price-history –instance-types m3.2xlarge –start-time 2017-07-25T07:08:09 –availability-zone eu-west-1c[/java]

Autre facteur de coût : le stockage EBS associé à chaque machine. Hadoop repose historiquement sur HDFS qui lui-même repose sur des disques configurés en JBOD. En environnement AWS EMR, le stockage EBS est utilisé pour stocker les données temporaires de nos phases de processing et également pour HDFS. Nous pouvons associer plusieurs volumes EBS à nos instances, ce qui aura pour effet de démultiplier les I/O pour l’accès à ces données. Quoi qu’il arrive à nos volumes EBS sur EMR, ils sont éphémères et non reconstructibles. C’est également le parti pris par HDFS avec sa gestion des réplicats permettant de s’affranchir de la perte d’un disque ou d’une machine.

Quelle est ma scaling policy ?

Atout considérable d’EMR, l’augmentation dynamique de la capacité de traitement du cluster. Nous pouvons augmenter la capacité du cluster (Scale Out) ou la diminuer (Scale In) en fonction de multiples métriques du cluster (métriques remontées par CloudWatch). À titre d’exemple, nous pouvons engendrer un Scale Out du cluster si la mémoire disponible dans YARN est inférieure à 15 % du total disponible. Les nœuds master quant à eux ne sont pas scalables, les politiques de scale In/Out sont uniquement réalisables sur les noeuds de type Core ou Task. CloudWatch va surveiller les métriques de YARN et agir sur EMR au travers du rôle AmazonElasticMapReduceAutoScaling.

La figure ci-dessous présente la stratégie par défaut pour l’Auto Scaling d’EMR.

Les stratégies de scaling sont cumulables. Il faut par conséquent veiller à ce qu’elles ne se mordent pas la queue !!

Dois-je adapter la configuration des services ?

Toutes les configurations par défaut des composants Hadoop sont modifiables dès l’instanciation du cluster. Par exemple, nous pouvons modifier les paramètres par défaut des services Hadoop tels que YARN ou Spark. Ces actions sont possibles directement dans l’aws cli avec l’utilisation de l’option --configuration ou la référence à un fichier json déposé au préalable sur S3. Une troisième méthode consiste à passer par la console pour modifier les options de configuration des services. Les différents niveaux de configuration des services Hadoop sont modifiables qu’il s’agisse d’un fichier hive-site.xml ou d’une proprité du fichier hue.ini.

Voici quelques exemples d’actions de configuration qu’il est bon de réaliser :

Premier exemple, configurer Hive pour qu’il utilise une base de données RDS MariaDB ainsi que quelques propriétés au sein du fichier hive-site.xml.

[java]{
"Classification": "hive-site",
"Properties": {
"hive.server2.authentication": "NOSASL",
"hive.enforce.bucketing": "true",
"hive.support.concurrency": "true",
"javax.jdo.option.ConnectionUserName": "<DB_USER>",
"hive.txn.manager": "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager",
"javax.jdo.option.ConnectionDriverName": "org.mariadb.jdbc.Driver",
"javax.jdo.option.ConnectionPassword": "<DB_PASSWORD>",
"hive.exec.dynamic.partition.mode": "nonstrict",
"javax.jdo.option.ConnectionURL": "jdbc:mysql://<DB_HOST>:3306/hivemetastoredb?createDatabaseIfNotExist=true",
"hive.compactor.initiator.on": "true",
"hive.compactor.worker.threads": "1"
},
"Configurations": []
}[/java]

Ceci aura pour effet de garantir la non perte des métadonnées des tables déclarées dans Hive dans le cas où le cluster disparaît.

Deuxième exemple, mais d’un niveau différent, configurer Hue pour qu’il utilise une base de données RDS MariaDB. Dans le précédent exemple, on configurait les propriétés d’un service Hadoop (Hive) dans la zone Properties. Ici, on modifie la zone [desktop] et sa sous zone [[database]] à l’aide du JSON suivant.

[java]{
"Classification": "hue-ini",
"Properties": {},
"Configurations": [
{
"Classification": "desktop",
"Properties": {},
"Configurations": [
{
"Classification": "database",
"Properties": {
"password": "<DB_PASSWORD>",
"engine": "mysql",
"port": "3306",
"host": "<DB_HOST>",
"name": "<DB_NAME>",
"user": "<DB_USER>"
},
"Configurations": []
}
]
}
]
}[/java]

Un troisième exemple, consiste à configurer Zeppelin pour qu’il stocke les Notebooks dans S3 :

[java]{
"Classification": "zeppelin-env",
"Properties": {},
"Configurations": [
{
"Classification": "export",
"Properties": {
"ZEPPELIN_NOTEBOOK_S3_BUCKET": "<BUCKET_S3>",
"ZEPPELIN_NOTEBOOK_S3_USER": "<USER_S3>",
"ZEPPELIN_NOTEBOOK_STORAGE": "org.apache.zeppelin.notebook.repo.S3NotebookRepo"
},
"Configurations": []
}
]
}[/java]

Grâce à cette configuration, les Notebooks développés sur Zeppelin seront sauvegardés dans S3.

Actions d’amorçage ?

Les actions d’amorçage sont lancées avant l’exécution des traitements sur le cluster et sont optionnelles. Elles permettent entre autre, d’installer des paquets supplémentaires sur les machines, d’ajouter des utilisateurs, … Un ensemble de scripts stockés sur S3 et appelables en tant qu’action d’amorçage est mis à disposition par AWS.

Let’s go

Nous sommes maintenant prêts pour démarrer un EMR . Cette action peut se faire de différentes manières : pour l’illustrer nous allons prendre un besoin de cluster EMR fictif détaillé dans le tableau ci-dessous :

fs.s3.consistent.retryPeriodSeconds à 10 secondes

fs.s3.consistent à true

fs.s3.consistent.retryCount à 5

fs.s3.consistent.metadata.tableName à

EmrFSMetadata

s3n://aws-logs-XXX-eu-west-1/elasticmapreduce/

La commande d’exemple suivante (à adapter) utilisant l’aws cli suffit à démarrer un cluster :

[java]aws emr create-cluster \
–applications Name=Hadoop Name=Hue Name=Spark Name=Hive Name=Zeppelin Name=HCatalog \
–ec2-attributes ‘{"KeyName":"bbo_emr","InstanceProfile":"EMR_EC2_DefaultRole","SubnetId":"subnet-XXX","EmrManagedSlaveSecurityGroup":"sg-XXXX","EmrManagedMasterSecurityGroup":"sg-XXX"}’ \
–release-label emr-5.7.0 \
–log-uri ‘s3n://aws-logs-XXX-eu-west-1/elasticmapreduce/’ \
–instance-groups ‘[{"InstanceCount":1,"BidPrice":"0.10","InstanceGroupType":"MASTER","InstanceType":"m3.xlarge","Name":"Groupe d instances maître – 1"},{"InstanceCount":2,"BidPrice":"0.10","InstanceGroupType":"CORE","InstanceType":"m3.xlarge","Name":"Groupe d instances principal – 2"}]’ \
–configurations ‘[{"Classification":"emrfs-site","Properties":{"fs.s3.consistent.retryPeriodSeconds":"10","fs.s3.consistent":"true","fs.s3.consistent.retryCount":"5","fs.s3.consistent.metadata.tableName":"EmrFSMetadata"},"Configurations":[]}]’ –auto-scaling-role EMR_AutoScaling_DefaultRole \
–ebs-root-volume-size 10 \
–service-role EMR_DefaultRole \
–enable-debugging \
–name ‘my-first-emr’ \
–scale-down-behavior TERMINATE_AT_INSTANCE_HOUR \
–region eu-west-1[/java]

Quand on connaît la multitude d’opérations à réaliser pour mettre en service un cluster Hadoop traditionnel, pouvoir le faire en une commande est exceptionnel. Une fois la commande exécutée, le cluster passe par les différents états décrits plus haut. L’état du cluster est visible via les différents moyens d’interfaçage avec AWS. Dans le cas d’un long running cluster, les interfaces utilisateur Zeppelin et Hue sont opérationnelles au bout d’une dizaine de minutes seulement !

Pour y accéder, deux possibilités :

• Connexion SSH au master et création de tunnels vers les différentes WebUI disponibles.
• Connexion SSH au master incluant un proxy socks avec dynamic port forwarding.

Le code de l’infrastructure EMR construite peut vivre au sein de vos projets Data et suivre ses propres releases. Il existe plusieurs possibilités pour créer, tester et déclencher le code associé à notre infrastructure.

Voici deux outils bien connus des profils Devops

• Terraform : Outil multi fournisseurs de Cloud développé par HashiCorp, il permet de planifier, provisionner et détruire notre infrastructure décrite dans des fichiers à l’extension .tf. Il est très complet mais il n’a pas nativement toutes les fonctionnalités AWS spécifiques à EMR. Heureusement, des JIRA sont en cours de résolution pour les prochaines versions et il est possible de développer ses propres composants. Un exemple de configuration pour EMR est fourni.

• CloudFormation : Intégré à AWS uniquement, il permet d’élaborer une infrastructure et de la décrire sous forme JSON ou YAML. Cette infrastructure « Cloud formée » peut être ensuite construite via les différents moyens d’interagir avec AWS, mais également avec un outil comme Ansible. De par sa disponibilité au sein de la constellation des services Amazon, CloudFormation interagit facilement avec EMR.

Flexible, facile et complet pour un besoin Hadoop rapide

EMR est une solution très flexible que l’on peut mettre facilement en service. Elle adresse un nombre important de cas d’utilisation et contient les dernières évolutions de l’écosystème Hadoop. Les avantages à l’utiliser sont très nombreux. En résumé, il n’y a pas de freins à l’utiliser excepté son prix et sa forte adhérence à AWS.

Sur du long terme, les coûts de fonctionnement peuvent être ramenés au strict minimum via l’utilisation des instances ponctuelles et des clusters éphémères.

Plusieurs alternatives à EMR existent. Elles feront l’objet d’articles dédiés. Parmi ces solutions, certaines permettent de réduire encore la facture de l’infrastructure Hadoop mais ne sont pas aussi flexibles qu’EMR et amènent une gestion fine des composants.