Initiation à l'écosystème Hadoop, tout ce que vous avez toujours voulu savoir sur Hive, Pig, Sqoop, Spark et leurs amis...

C'est pour combler ces lacunes qu'un ensemble de technologies regroupées sous le nom d'écosystème Hadoop a été développé. L'écosystème Hadoop fournit une collection d'outils et technologies spécialement conçus pour faciliter le développement, le déploiement et le support des solutions Big Data. L'objectif de cet article est double :

- passer en revue la fonction de chacun de ces outils de l'écosystème.
- annoncer la sortie prochaine (prévue pour Mars/Avril) de notre ouvrage intitulé " Maîtrisez l'utilisation de l'écosystème Hadoop : Initiation à l'écosystème Hadoop " chez les éditions Eyrolles qui les traite de façon exhaustive.

L'ouvrage a pour but d'aider tout professionnel à utiliser de façon professionnelle 18 technologies clé de l'écosystème Hadoop à savoir : Spark, Hive, Pig, Impala, ElasticSearch, HBase, Lucene, HAWQ, MapReduce, Mahout, HAMA, TEZ, Phoenix, YARN, ZooKeeper, Storm, Oozie, et Sqoop. A la fin de sa lecture, vous serez également capable de choisir les solutions techniques adaptées à vos problèmes de Big Data.

En attendant la sortie officielle de l'ouvrage, vous pouvez vous procurer gratuitement un extrait portant sur ElasticSearch. Pour le recevoir directement dans votre boîte mail (30 pages), cliquez sur le lien suivant : http://www.data-transitionnumerique.com/extrait-ecosystme-hadoop/

La carte heuristique suivante présente de façon globale l'écosystème Hadoop et résume particulièrement bien la structure de l'ouvrage.

Maintenant passons au vif du sujet. La configuration de base de l'écosystème Hadoop contient les technologies suivantes : Spark, Hive, PIG, HBase, Sqoop, Storm, ZooKeeper et Oozie.

Avant d'expliquer ce que c'est que Spark, rappelons que pour qu'un algorithme puisse s'exécuter sur plusieurs nœuds d'un cluster Hadoop, il faut qu'il soit parallélisable. Ainsi, on dit d'un algorithme qu'il est "scalable" s'il est parallélisable (et peut donc profiter de la scalabilité d'un cluster). Hadoop est une implémentation du modèle de calcul MapReduce. Le problème avec le MapReduce est qu'il est bâti sur un modèle de Graphe Acyclique Direct. En d'autres termes, l'enchaînement des opérations du MapReduce s'exécutent en trois phases séquentielles directes et sans détour (Map -> Shuffle -> Reduce), aucune phase n'est itérative (ou cyclique). Le modèle acyclique direct n'est pas adapté à certaines applications, notamment celles qui réutilisent les données à travers de multiples opérations, telles que la plupart des algorithmes d'apprentissage statistique, itératifs pour la plupart, et les requêtes interactives d'analyse de données. Spark est une réponse à ces limites, c'est un moteur de calcul qui effectue des traitements distribués en mémoire sur un cluster. Autrement dit, c'est un moteur de calcul in-memory distribué. Comparativement au MapReduce qui fonctionne en mode batch, le modèle de calcul de Spark fonctionne en mode interactif, c'est à dire, monte les données en mémoire avant de les traiter et est de ce fait très adapté au traitement de Machine Learning.

Hive est une infrastructure informatique similaire au Data Warehouse qui fournit des services de requêtes et d'agrégation de très gros volumes de données stockées sur un système de fichier distribué de type HDFS. Hive fournit un langage de requête basé sur le SQL (norme ANSI-92) appelé HiveQL (Hive Query Language), qui est utilisé pour adresser des requêtes aux données stockées sur le HDFS. Le HiveQL permet également aux utilisateurs avancés/développeurs d'intégrer des fonctions Map et Reduce directement à leurs requêtes pour couvrir une plus large palette de problèmes de gestion de données. Lorsque vous écrivez une requête en HiveQL, cette requête est transformée en job MapReduce et soumis au JobTracker pour exécution par Hive. Voici un exemple de requête écrite en HiveQL. Trouver la température maximale par année :

USE default ;
CREATE TABLE records (year string, temperature INT, quality INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' ;
LOAD DATA LOCAL 'data/sample.txt' OVERWRITE INTO TABLE records ;
SELECT year, MAX(temperature) FROM records WHERE temperature !=9999 AND (quality == 0 OR quality == 1) GROUP BY year ;

Pig est un environnement d'exécution de flux interactifs de données sous Hadoop. Il est composé de 2 éléments :
- un langage d'expression de flux de données appelé le Pig Latin ;
- et un environnement Interactif d'exécution de ces flux de données ;
Le langage offert par Pig, le Pig Latin, est à peu près similaire au langage de Scripting tels que Perl, Python, ou Ruby. Cependant, il est plus spécifique que ces derniers et se décrit mieux sur le terme "langage de flux de données" (data flow language). Il permet d'écrire des requêtes sous forme de flux séquentiels de données source pour obtenir des données " cible " sous Hadoop à la façon d'un ETL. Ces flux sont ensuite transformés en fonctions MapReduce qui sont enfin soumises au jobtracker pour exécution. Pour faire simple, Pig c'est l'ETL d'Hadoop. Programmer en Pig Latin revient à décrire sous forme de flux indépendants mais imbriqués, la façon dont les données sont chargées, transformées, et agrégées à l'aide d'instructions Pig spécifiques appelées opérateurs. La maîtrise de ces opérateurs est la clé de la maîtrise de la programmation en Pig Latin, d'autant plus qu'ils ne sont pas nombreux relativement au Hive par exemple. L'exemple précédent en HiveQL donne ceci en Pig Latin :

records = LOAD 'data/samples.txt AS (year: chararray, temperature : int, quality: int);
filtered_records = FILTER records BY temperature !=9999 AND (quality ==0 OR quality == 4);
grouped_records = GROUP filtered_records BY year ;
Max_temp = FOREACH grouped_records GENERATE group, MAX filtered_records.temperature)
DUMP max_temp ;

Avant de parler de HBase, nous allons rappeler que les SGBDR, qui sont jusqu'à présent utilisés pour la gestion des données ont montré très rapidement leurs limites face d'une part la forte volumétrie des données et d'autre part face à la diversité des données. En effet, les SGBDR sont conçus pour gérer uniquement des données structurées (table de données en ligne/colonnes), de plus l'augmentation du volume des données augmente le temps de latence des requêtes. Cette latence est préjudiciable dans le cadre de nombreux métiers requérant des réponses en temps quasi-réel. Pour répondre à ces limites, de nouveaux SGBD dit "NoSQL" ont vu le jour. Ceux-ci n'imposent pas de structure particulière aux données, sont capables de distribuer le stockage et la gestion des données sur plusieurs nœuds et sont scalables. A titre de rappel, la scalabilité signifie que la performance du système reste stable avec l'augmentation de la charge de traitement. HBase fait partie de cette catégorie de SGBD.
HBase est un SGBD distribué, orienté-colonne qui fournit l'accès en temps réel aussi bien en lecture qu'en écriture aux données stockées sur le HDFS. Là où le HDFS fournit un accès séquentiel au données en batch, non-approprié pour des problématiques d'accès rapide à la donnée comme le Streaming, HBase couvre ces lacunes et offre un accès rapide aux données stockées sur le HDFS.
Il a été conçu à partir du SGBD de Google "Big Table" et est capable de stocker de très grosses volumétries de données (milliard de lignes/colonnes). Il dépend de ZooKeeper, un service de coordination distribuée pour le développement d'applications.

Sqoop ou SQL-to-Hadoop est un outil qui permet de transférer les données d'une base de données relationnelle au HDFS d'Hadoop et vice-verça. Il est intégré à l'écosystème Hadoop et est ce que nous appelons le planificateur d'ingestion des données dans Hadoop. Vous pouvez utiliser Sqoop pour importer des données des SGBDR tels que MySQL, Oracle, ou SQL Server au HDFS, transformer les données dans Hadoop via le MapReduce ou un autre modèle de calcul, et les exporter en retour dans le SGBDR. Nous l'appelons planificateur d'ingestion des données parce que tout comme Oozie (plus bas), il automatise ce processus d'import/export et en planifie le moment d'exécution. Tout ce que vous avez à faire en tant qu'utilisateur c'est d'écrire les requêtes SQL qui vont être utilisées pour effectuer le mouvement d'import/export. Par ailleurs, Sqoop, utilise le MapReduce pour importer et exporter les données, ce qui efficace et tolérant aux pannes. La figure suivante illustre particulièrement bien les fonctions de Sqoop.

Pour comprendre Storm, il faut comprendre la notion d'architectures lambda (λ) et pour comprendre l'intérêt des architectures lambda, il faut comprendre le concept d'objets connectés. Les objets connectés ou Internet des objets (IoT - Internet of Things en anglais) représente l'extension d'Internet à nos vies quotidiennes. Elle génère des données en streaming et dans la plupart de ses problématiques, nécessite que les données soient traitées en temps réel. Les modèles que vous connaissez tels que les modèles de calcul Batch ne sont pas adaptés aux problématiques temps réel que soulève l'IoT. Même les modèles de calcul interactif ne sont pas adaptés pour faire du traitement continu en temps réel. A la différence des données opérationnelles produites par les systèmes opérationnels d'une entreprise comme la finance, le marketing, qui même lorsqu'elles sont produites en streaming peuvent être historisées pour un traitement ultérieur, les données produites en streaming dans le cadre des phénomènes comme l'IoT ou Internet se périment (ou ne sont plus valides) dans les instants qui suivent leur création et exigent donc un traitement immédiat. En dehors des objets connectés, les problématiques métier comme la lutte contre la fraude, l'analyse des données de réseau sociaux, la géolocalisation, exigent des temps de réponse très faibles, quasiment de l'ordre de moins d'une seconde.
Pour résoudre cette problématique dans un contexte Big Data, des architectures dites λ ont été mises sur pieds. Ces architectures ajoutent au MapReduce 2 couches de traitements supplémentaires pour la réduction des temps de latence. Storm est une implémentation logicielle de l'architecture λ. Il permet de développer sous Hadoop des applications qui traitent les données en temps réel (ou presque).

La synchronisation ou coordination de la communication entre les nœuds lors de l'exécution des tâches parallèles est l'un des problèmes les plus difficiles dans le développement d'application distribuée. Pour résoudre ce problème, Hadoop a introduit dans son écosystème des outils dits de coordination de service, en l'occurrence ZooKeeper. ZooKeeper prend en charge la complexité inhérente de la synchronisation de l'exécution des tâches distribuées dans le cluster et permet aux autres outils de l'écosystème Hadoop de ne pas avoir à gérer ce problème eux-mêmes. Il permet également aux utilisateurs de pouvoir développer des applications distribuées sans être des experts de la programmation distribuée. Sans entrer dans les détails complexes de la coordination des données entre les nœuds d'un cluster Hadoop, ZooKeeper fournit un service de configuration distribué, un service de distribution et un registre de nommage pour les applications distribuées. ZooKeeper est le moyen utilisé par Hadoop pour coordonner les jobs distribués.

Par défaut, Hadoop exécute les jobs au fur et à mesure qu'ils sont soumis par l'utilisateur sans tenir compte de la relation qu'ils peuvent avoir les uns avec les autres. Or, les problématiques pour lesquelles l'on utilise Hadoop demandent généralement la rédaction d'un ou de plusieurs jobs complexes. Lorsque les 2 jobs seront soumis au JobTracker (ou à YARN) par exemple, celui-ci va les exécuter sans faire attention au lien qui existe entre eux, ce qui risque de causer une erreur (exception) et entraîner l'arrêt du code. Comment fait-on pour gérer l'exécution de plusieurs jobs qui sont relatifs au même problème ? Pour gérer ce type de problème, la solution la plus simple actuellement consiste à utiliser un planificateur de jobs, en l'occurrence Oozie. Oozie est un planificateur d'exécution des jobs qui fonctionne comme un service sur un cluster Hadoop. Il est utilisé pour la planification des jobs Hadoop, et plus généralement pour la planification de l'exécution de l'ensemble des jobs qui peuvent s'exécuter sur un cluster, par exemple un script Hive, un job MapReduce, un job Hama, un job Storm, etc. Il a été conçu pour gérer l'exécution immédiate, ou différée de milliers de jobs interdépendants sur un cluster Hadoop automatiquement. Pour utiliser Oozie, il suffit de configurer 2 fichiers XML : un fichier de configuration du moteur Oozie et un fichier de configuration du workflow des jobs.

Bien évidemment, vous l'aurez compris, il est impossible de développer de façon exhaustive toutes ces technologies de l'écosystème Hadoop dans un article. Dans notre ouvrage " Maîtrisez l'utilisation de l'écosystème Hadoop : Initiation à l'écosystème Hadoop " à paraître, nous les couvrons toutes dans leur intégralité. Cliquez sur le lien suivant pour être tenu informés de sa sortie et avoir en exclusivité l'étude d'une technologie de l'écosystème Hadoop abordée également dans l'ouvrage - ElasticSearch : http://www.data-transitionnumerique.com/extrait-ecosystme-hadoop/