LE SQL DANS HADOOP : Les moteurs natifs SQL sur Hadoop

Dans le tutoriel précédent : le SQL dans Hadoop – Hive & Pig, vous avez appris comment utiliser une abstraction pour exécuter des instructions SQL compatibles avec la norme ANSI 92 telle que Apache Hive sur Hadoop. Dans ce tutoriel, nous allons aller plus loin. Nous allons vous montrer comment certains moteurs SQL exécutent directement et nativement du SQL dans Hadoop.

Hadoop est en train de gagner du terrain dans les entreprises, celles-ci utilisent de plus en plus le HDFS comme répertoire de stockage central pour toutes leurs données ; données provenant de plusieurs sources telles que les systèmes opérationnels (comptabilité, marketing, finance, Ressources Humaines, etc.), les médias sociaux, Internet, les données de capteurs, et les données de diverses applications. Au même moment, la majorité des outils de gestion et d’exploitation des données utilisées par celles-ci (par exemple l’ERP SAP, Oracle, SAS, Tableau, Excel, etc.) s’appuient sur le SQL. Les analystes métiers et autres professionnels qui travaillent sur ces outils sont à l’aise avec le SQL, et celui-ci leur est très familier. C’est l’une des raisons majeures qui a provoqué le développement des offres SQL sur Hadoop ces dernières années. Facebook a été le premier sur le marché à offrir une solution SQL sur Hadoop, Hive avec un langage d’interrogation similaire au SQL, le HiveQL. Hive a permis de transformer et de compiler les requêtes HiveQL en série de job MapReduce. A mesure que les besoins des utilisateurs ont évolués et que les exigences du Numérique se sont accrues, le temps de latence de toutes les solutions SQL sur Hadoop qui transforment les requêtes en Job MapReduce (les langages d’abstraction) sont devenus inacceptables. Pour résoudre ce problème, la majorité des éditeurs ont décidé de développer des moteurs SQL natifs, massivement parallèles qui exécutent directement le SQL sur le HDFS sans passer par le MapReduce. Vous vous demandez peut-être comment cela se fait-il qu’une solution soit capable d’exécuter nativement le SQL directement sur le HDFS ? En fait, le concept de moteur SQL massivement parallèle n’est pas nouveau. Certains systèmes de gestion des bases de données relationnelles tels que GreenPlum, PostgreSQL, Informix XPS, IBM DB2 UDB, NCR Teradata, SQL Server Parallel DWH, ou Sybase IQ12-Multiplex, distribuent les bases de données sur un ensemble d’ordinateurs connectés à un réseau LAN et exécutent des requêtes SQL de façon concurrente (parallèle) et simultanée sur ces machines. Parce qu’ils sont capable d’exécuter du SQL sur des architectures distribuées, ces systèmes ont été qualifiés de Systèmes Parallèles de Gestion de Bases de données relationnelles ou de Bases de données à architecture MPP (Massively Parallel Processing). Couramment, on les appelle les Bases de Données Parallèles (Parallel Data Base ou Parallel DataWarehouse) ou Massively Parallel Processing Databases (MPP DB). Etant donné qu’un cluster Hadoop est par définition Shared-Nothing (et donc MPP), tous les moteurs SQL natifs Hadoop se sont simplement inspirés des principes des bases de données parallèles. Dans cette partie, vous allez apprendre d’une part les principes de fonctionnement des moteurs SQL sur Hadoop, et d’autre part l’un des moteurs SQL sur Hadoop natifs les plus connus : Impala. Etant donné que les principes de tous ces moteurs sont sensiblement les mêmes comme on va le voir, la connaissance d’Impala vous permettra de pouvoir investiguer vous-même d’autres moteurs SQL Hadoop sans aucun problème. Mais pour que vous ayez la meilleure compréhension possible des moteurs SQL sur Hadoop, il vous faut obligatoirement comprendre le fonctionnement des bases de données parallèles (Massively Parallel Database), tout commence par là.

1.1. Fonctionnement des bases de données parallèles (MPP DB)

Dans les architectures Client/serveur, la seule forme de distribution consiste à répartir les charges de calcul entre une machine serveur et plusieurs machines clientes. Malgré tout, elle reste une approche centralisée, car la machine Serveur centralise les tâches d’administration et répond à toutes les requêtes. Pour faire du traitement massivement parallèle, il faut utiliser un cluster et choisir entre 3 types d’architectures : les architectures Shared-Disk, les architectures Shared-Memory et les architectures Shared-Nothing. En mi-décembre 1999, l’entreprise Deutsche Telecom a annoncé que son DataWarehouse stockait 100 Téraoctets de données (soit 102 400 Go). La question qui se pose alors est : comment faire pour qu’un SGBDR qui héberge un DataWarehouse d’une telle volumétrie réponde efficacement aux requêtes SQL des utilisateurs ?

1.1.1. Architecture des bases de données parallèles

Avec la baisse des coûts d’ordinateurs, l’augmentation de la puissance de calcul et le développement du réseau, l’approche utilisée pour résoudre un problème comme celui de Deutsche Telecom consiste à utiliser un cluster de machines à architecture Shared-Memory ou Shared-Nothing, à répliquer dans chaque nœud du cluster une structure de base de données identique et vide, ensuite à partitionner les données de la base de données et enfin à distribuer les partitions à tous les nœuds contenant la réplique de la structure de la base de données. Avec ce mode de fonctionnement, les requêtes SQL qui arrivent dans le système sont parallélisées et exécutées de façon concurrente comme des processus indépendants sur les nœuds contenant les partitions de base de données. Comme chaque nœud contient une partie de la base de données, il s’en suit que la charge de calcul (les 100 To par exemple) est répartie à travers tout le cluster et que les processus de la requête s’exécutent directement In Situ (en local) dans le noeud. C’est cela le principe des bases de données parallèles. Pour des besoins de scalabilité, les SGBDR parallèles de DataWarehouse privilégient l’architecture Shared-Nothing. C’est d’ailleurs pourquoi on les appelle MPP Database (Massively Parallel Processing database) pour faire référence au fait que dans une architecture Shared-Nothing, chaque nœud constitue un processeur indépendant, par conséquent l’ensemble des nœuds du cluster forme un cluster de processeurs.
Dans les bases de données parallèles, la structure de la base de données est répliquée entre tous les nœuds d’un cluster Shared-Nothing et ses données sont partitionnées et distribuées à travers ces nœuds, de sorte que chaque nœud héberge une partition de la base de données. Toute communication dans le cluster se fait via le réseau (LAN ou fibre optique). Par exemple, si la base de données contient une table clients et que le cluster est constitué de 50 nœuds, alors il y’aura 50 instances de la table clients, et chacune aura une partition différente des données de la table. Pour exécuter une requête SQL sur cette table, chaque nœud va simultanément exécuter la requête sur sa base de données locale. Lorsque l’exécution de la requête sera terminée dans tous les nœuds, une phase de jointure va se déclencher dans le nœud central du cluster et va agréger les résultats de la requête.
Le problème majeur avec l’approche de base de données parallèle est la localisation des données. Lorsque la base de données est installée dans une architecture Shared-Memory, il n’y’a qu’une copie de la base de données, accessible à tous les nœuds. Dans une architecture Shared-Nothing par contre, comme chaque nœud possède sa propre mémoire et son propre disque dur, la base de données doit être partitionnée et distribuée à tous les nœuds du cluster. Chaque nœud ne peut accéder qu’à sa propre partition de base de données. Par conséquent, la façon dont la base de données est partitionnée a un impact direct sur le niveau de complexité des programmes qui vont exploiter cette base de données, spécialement pour les opérations de jointure. Souvenez vous que le traitement parallèle en architecture Shared-Nothing exige que les taches soient totalement indépendantes. Le partitionnement de la base de données doit donc se faire de façon à ce que les partitions soient les plus indépendantes possibles. A quelques exceptions prêtes, tous les systèmes de bases de données parallèles utilisent le partitionnement horizontal pour diviser les tables de la base de données maître en partitions. Le partitionnement horizontal consiste à diviser une table par lignes, il se différencie du partitionnement vertical qui lui consiste à partitionner une table par colonne. Le partitionnement vertical est plus approprié lorsque la problématique à traiter implique de nombreuses jointures, mais très coûteux quand il s’agit de traitement sur lignes. Avec le partitionnement horizontal, comme la table est partitionnée en lignes et que les lignes sont complètement indépendantes les unes des autres, les partitions peuvent être traitées de façon parallèle.
Maintenant, une fois qu’on a les L partitions de lignes de table, comment décide-t-on de l’affectation de chaque partition aux nœuds ? Il y’a 3 méthodes d’affectation de partition aux nœuds : l’affectation par intervalle, l’affectation Round-robin et l’affectation par hachage (Hashing).
– Dans l’affectation par intervalle, une ou plusieurs colonnes de la table sont spécifiées comme clé de partitionnement. Généralement c’est la clé primaire de la table qui est utilisée comme clé de partitionnement. Chaque nœud reçoit un intervalle de valeurs de la clé. L’avantage avec cette méthode est que la clé agit comme index. De plus, la clé assure une répartition plus ou moins équilibrée des lignes entre les nœuds. Cependant pour obtenir ces deux avantages, il faut obligatoirement que les valeurs de la clé soient uniformes et le restent dans le temps. Sinon, les partitions n’auront pas la même taille, ce qui va déséquilibre les charges de calcul dans le cluster ;
– Pour résoudre le problème de non-uniformité de la clé et le déséquilibre que cela peut engendrer dans la répartition des charges de calcul, l’affection round-robin assigne la ligne suivante au noeud suivant, indépendamment de la valeur de la clé. Cela assure que les lignes soient aussi distribuées que possible. L’inconvénient cependant est qu’on ne sait plus sur quel disque une ligne particulière est située ;
– L’affectation par hachage est une combinaison des deux méthodes précédentes. Comme dans l’affectation Round-robin, elle évite les déséquilibres de charge en utilisant une fonction hash sur la clé pour randomiser les lignes et obtenir des partitions suffisamment égales. Comme dans l’affectation par intervalle, le hachage assure également que les lignes qui ont une valeur de clé identique seront placées dans la même partition et donc affectées au même nœud (co-localisation des données de même valeur de clé). Ainsi, les jointures seront plus simples à réaliser et il n’y’aura pas besoin de déplacement de données dans le cluster ;
Nous allons illustrer concrètement tous ces principes que nous venons de voir avec GreenPlum, un SGBDR MPP développé par Pivotal. Considérons la base de données parallèle contenant la table products suivante :

Supposons que la base de données est parallélisée sur un cluster de 3 nœuds. La figure suivante illustre l’architecture d’un SGBDR MPP classique.

La distribution des données se fait par partitionnement horizontal des données des tables de la base de données parallèle. Dans GreenPlum, cela s’accomplit lors de la création de la table avec le SQL par l’instruction DISTRIBUTED BY. GreenPlum propose 2 modes d’affectation des partitions aux nœuds de données : l’affection round-robin et l’affectation par hachage. Dans cet exemple, nous allons utiliser l’affection par hachage. Nous allons également utiliser comme clé de partitionnement product_id. la requête de création de la table products est la suivante :

CREATE TABLE products (

   product_id       INTEGER,

   product_name     varchar(200),

   price            DOUBLE

) DISTRIBUTED BY (product_id) ;

Une fois que la table a été crée, nous pouvons maintenant insérer les 13 lignes de la table et voir l’effet du hachage sur l’affectation des lignes aux nœuds de données.

INSERT INTO products VALUES (1001, “Tablette Amazon Kindle”, 56);

INSERT INTO products VALUES (1002, “Iphone Apple 5S Nano”, 400);

                 …… …… …… …… …… … ……

INSERT INTO products VALUES (1012, “Costume Class”, 195);

La figure suivante illustre le partitionnement et la distribution des données dans la base parallèle.

Remarquez que les tables partitionnées sont bien réparties (elles ont toute la même taille) et que les valeurs sont distribuées de façon aléatoire. Ce sont là les caractéristiques de l’affectation par hachage. Si l’affectation s’était faite par exemple par intervalle, les lignes auraient été partitionnées par valeur consécutives, et n’auraient pas nécessairement été aussi bien réparties. Nous allons maintenant passer au deuxième aspect des bases parallèles : l’exécution des requêtes SQL.

1.1.2. Exécution des requêtes SQL dans les bases parallèles

Les SGBDR s’interrogent avec un langage bien connu, qui sert aussi bien de langage d’interrogation que de langage de définition de données : le SQL (Structured Query Language). Tout comme le HiveQL, le SQL permet à l’utilisateur de décrire les opérations à effectuer sur la base de données à l’aide d’un ensemble de clauses. Le SQL est le standard en matière d’interrogation des bases de données relationnelles et par sa syntaxe simple, est devenu depuis, le langage favori des analystes métiers. En réalité, le SQL fonctionne également comme une abstraction, elle permet à l’utilisateur de décrire le QUOI sans décrire le COMMENT. Les requêtes SQL ne sont pas directement exécutées, elles passent toutes par trois phases :

• Premièrement, leur syntaxe est vérifiée par le système. Si l’utilisateur a mal saisi le code ou s’est trompé sur l’écriture d’une clause, alors le système génère une erreur et le programme s’arrête ;
• Deuxièmement, lorsque la syntaxe de la requête est validée, la requête SQL est transformée en plan d’exécution, un arbre de décision hiérarchique similaire à un graphe acyclique direct qui indique au moteur du SGBDR l’ordre dans lequel les clauses de la requête doivent être exécutées ;
• Dernièrement, le plan d’exécution est compilé en instructions machine et ce sont ces instructions qui sont exécutées par le processeur de la machine sur laquelle le SGBDR est installé. La figure suivante illustre le processus classique d’exécution d’une requête SQL ;

Attention, dans un contexte distribué en général et un SGBDR Parallèle en particulier par contre, le processus général est le même, mais le schéma d’exécution n’est pas le même. Les requêtes SQL sont décomposées en processus parallèles, et exécutées de façon concurrente sur tous les nœuds contenant une copie de la base de données parallèle. Attention, ce sont les mêmes instructions (la même requête SQL) qui sont exécutées simultanément sur les nœuds du cluster. La machine centrale et les nœuds de calcul sont impliqués dans l’exécution de la requête. La machine centrale génère le plan d’exécution de la requête SQL et la répartit aux nœuds de calcul contenant une partition de la base de données, tandis que les nœuds de calcul la compile et l’exécute. De façon générale, la requête suit le schéma décrit par la figure suivante :

En fonction du type de requête, le plan d’exécution n’est pas le même. Nous allons illustrer cette figure avec trois exemples. Reprenons le cas précédent et considérons que nous voulons réaliser une projection (opération de sélection en SQL) du produit d’id = 10005. La requête SQL qui permet de faire cela est la suivante :

SELECT * FROM products WHERE poduct_id = 10005;

Regardons comment cette requête s’exécute dans la base de données parallèle et l’impact du partitionnement de la table sur la performance de la requête. Lorsque cette requête est lancée, elle est transformée en plan d’exécution par le nœud central du cluster. Etant donné que cette requête est un filtre sur la colonne qui sert de clé de partitionnement, le plan d’exécution va être envoyé uniquement sur le nœud contenant la valeur 1005 (le nœud 2 dans notre exemple) et celui-ci va la compiler, l’exécuter sur sa partition et renvoyer le résultat à la machine centrale. Vous voyez que la performance est plus élevée que si la base était partagée en mémoire (comme dans les architectures centralisées ou les architectures Shared-Memory). La machine centrale est capable de déclencher le nombre exact de processus SQL qu’il faut parce que comme dans le nœud de référence d’Hadoop, il stocke les métadonnées concernant la distribution de la base de données. En d’autres termes, la machine centrale connait où sont localisées toutes les partitions des tables de la base de données et leur valeur. Supposons maintenant que nous voulons sélectionner les produits : 10001, 10002, 10003, 10004, 10005 et 10006. La requête correspondante est la suivante :

SELECT * FROM products WHERE poduct_id in (10001, 10006) ;

Lorsque cette requête est exécutée, la machine centrale la transforme en plan d’exécution, ensuite elle identifie dans ses métadonnées la localisation de chacune des valeurs du filtre et déclenche N processus d’exécution égal aux N partitions/nœuds où sont stockées ces données (dans notre exemple, les 3 nœuds de calcul). Les 3 processus vont s’exécuter en parallèle sur les trois nœuds contenant les partitions et les résultats seront transférés vers le Master, qui va les agréger pour obtenir la table de résultat finale. Vous voyez une fois de plus que l’exécution parallèle augmente drastiquement la performance des requêtes et permet de tirer pleinement profit des machines du cluster.
Pour le troisième cas, supposons que nous voulons effectuer une jointure entre deux tables. La figure 39 ci-après illustre une base de données parallèle composée de deux tables : la table products et la table orders. Chaque nœud aura une réplique de la structure de la base de données et une partition des deux tables telles qu’illustrer par la figure. Si les deux tables à joindre sont partitionnées selon la même clé, alors la jointure va se faire localement sur chaque nœud, ensuite les résultats de chacun seront transférés vers la machine centrale qui va les agréger. Si les deux tables ne sont pas distribuées selon la même clé, le SGBD Parallèle va utiliser les statistiques de la table pour copier les données de la plus petite table pour la placer dans le nœud de la plus grande pour effectuer la jointure. L’avantage avec les SGBDR MPP est que si les jointures deviennent coûteuses, alors vous pouvez simplement augmenter la performance du cluster en y augmentant le nombre de nœuds de calcul.

Voici, maintenant que vous connaissez en détail le fonctionnement des SGDBR Massivement parallèles, vous êtes prêts pour comprendre le fonctionnement des moteurs SQL MPP sur Hadoop.

1.2. Fonctionnement des moteurs natif SQL sur Hadoop

Originellement, le SQL est le langage des moteurs de bases de données relationnelles. Les bases de données relationnelles elles-mêmes tirent leur origine de la théorie de l’algèbre relationnelle et ont été mises au point par Edgar Frank CODD. Lorsqu’ Edgard Franck CODD créait le SQL, c’était pour fournir aux utilisateurs un langage déclaratif permettant de définir des opérations relationnelles (les tables et les vues), de les interroger (projections, groupes, filtres, jointures) et d’être interprété par des moteurs relationnels à l’exemple de SQL Server, DB2, Oracle. Il est très important que vous compreniez que le SQL est un langage de définition et d’interrogation des tables (données structurées en lignes colonnes) et que les requêtes SQL ne peuvent s’exécuter que par des moteurs de bases de données relationnelles. Les expressions « Moteur SQL sur Hadoop » ou « Moteur natifs SQL sur Hadoop » ne font pas référence à des moteurs relationnels distribués sur Hadoop, mais à des systèmes qui sont capable de fournir une couche de programmation SQL telle que spécifiée par la norme ANSI (SQL ANSI 92, SQL ANSI 2003, etc.) pour la spécification des travaux de traitement de données et fournir un moteur d’exécution de ces travaux directement sur les données stockées dans un cluster Hadoop. Hive a été le tout premier du genre. A la différence de Hive, ces systèmes ne transforment pas les requêtes SQL en jobs MapReduce, ni en un quelconque autre modèle de calcul distribué. Ils ne présentent pas juste une interface SQL. Ils exécutent directement le plan d’exécution des requêtes SQL de façon parallèle dans le cluster Hadoop. Comme nous l’avons vu plus haut, vous savez que le SQL ne s’exécute pas tel qu’il a été écrit par l’utilisateur, il est transformé en plan d’exécution et c’est ce plan qui est compilé et exécuté. Les moteurs SQL Natifs Hadoop exécutent directement le plan d’exécution des requêtes SQL sur le cluster Hadoop. C’est pourquoi on dit d’eux qu’ils sont « natifs SQL ». Le but qui est à l’origine de ces systèmes est de connecter les applications SQL existantes dans les systèmes décisionnels des entreprises à la puissance de calcul d’Hadoop, et ainsi de faciliter l’adoption d’Hadoop par les utilisateurs métiers et les entreprises.
Nous allons vous présenter les principes et les choix qui sont à la base d’un moteur SQL sur Hadoop et prendre l’exemple d’un d’entre eux : Impala. Actuellement, le marché regorge de plusieurs moteurs SQL sur Hadoop : Cloudera Impala, HadoopDB, Google Dremel, Apache Phoenix, Apache HAWQ, Apache Drill, Facebook Presto, VectorH, Vortex ou encore Spark SQL. Pour pouvoir exécuter de façon massivement parallèle du SQL sur un cluster Hadoop, les moteurs SQL sur Hadoop se sont inspirés du fonctionnement des SGBDR MPP et des principes des bases de données parallèles. Par exemple, le code source d’HadoopDB est constitué en majorité du code source de PostgreSQL, celui de HAWQ en majorité de GreenPlum, et Vortex celui de Vectorwise. Par contre, la ressemblance entre les deux types de systèmes s’arrête là. On note trois différences majeures entre les deux :

• Là où les SGBDR MPP doivent implémenter des mécanismes propriétaires de tolérance aux pannes, de réplication et de haute disponibilité, les moteurs SQL sur Hadoop s’appuient directement sur le HDFS pour fournir la haute disponibilité et la tolérance aux pannes nécessaires au traitement distribué ;
• Les SGDR MPP s’exécutent uniquement sur des tables relationnelles, tandis que les moteurs SQL sur Hadoop s’exécutent sur les données stockées dans le HDFS. Ces données peuvent être structurées (tables relationnelles, fichiers plats CSV) ou pas (fichiers JSON, XML, etc.). Là où les SGBDR MPP ont un contrôle sur les tables qu’ils manipulent, les moteurs SQL sur Hadoop eux n’en ont quasiment pas, des fichiers peuvent s’ajouter, être modifiés ou être supprimés du HDFS sans notification au moteur SQL sur Hadoop, compliquant ainsi l’optimisation des requêtes ;
• Les SGBDR MPP sont plus couteux financièrement qu’Hadoop. Coûteux financièrement en terme de logiciel, de matériel, d’administration, et même en terme de compétences (les développeurs et administrateurs de SGBDR MPP sont très rares). Hadoop par contre est open source et est en train de devenir le standard du traitement massivement parallèle. Acquérir des compétences sur Hadoop se vulgarise de plus en plus ;

Deux aspects gouvernent l’implémentation d’un moteur SQL MPP sur Hadoop. Les différences entre les solutions SQL sur Hadoop proviennent essentiellement de l’arbitrage entre les options offertes par ces deux aspects : le mode de stockage et la structure des données.

– Le mode de stockage

Le premier aspect le plus important des moteurs SQL sur Hadoop est leur mode de stockage. Paralléliser l’exécution des requêtes SQL sur les nœuds d’un cluster suppose que les traitements se fassent In Situ, c’est-à-dire en local sur chaque nœud. Dans un SGBDR Parallèle, les traitements sont faits In Situ par la réplication de la structure de la base de données et de ses tables aux nœuds de calcul du cluster. Dans un cluster Hadoop c’est le HDFS qui assure ce rôle de réplication. Le HDFS est le composant qui assure le traitement In Situ dans un cluster Hadoop. Pour faire le traitement In Situ, les éditeurs de moteurs SQL sur Hadoop ont le choix entre 3 options : soit ils s’appuient directement sur le HDFS pour le stockage, soit ils stockent les données dans les nœuds de calcul, sur un système de stockage différent du HDFS (par exemple un autre SGBD), soit alors ils transforment dynamiquement les données du HDFS en données propriétaires (par exemple en table relationnelle, en table HBase, etc.). Ce choix aura un impact sur la façon dont le moteur va exécuter les requêtes (exécution par le moteur, exécution par le SGBD, ou exécution partagée entre le SGBD et le moteur) et sur les SGBD auxquels le moteur pourra se connecter.

– La structure des données

Le deuxième aspect qui gouverne le développement d’un moteur SQL sur Hadoop est la structure des données. Dans les SGBDR Parallèles, toutes les données sont stockées sous forme de tables. Dans un cluster Hadoop par contre, les données sont stockées dans le HDFS, sous des formes complexes comme le JSON, XML, RDF, HTML ou sous forme de fichiers plats. Les éditeurs ici ont le choix entre trois options : soit ils supportent les formats natifs HDFS et n’imposent pas de format propriétaire, soit ils imposent un format spécifique (généralement propriétaire), soit alors ils transforment les données de leur structure de base en un format spécifique. Rare sont les moteurs SQL sur Hadoop qui, comme Apache Drill, sont capables d’exécuter du SQL sur des formats semi-structurés (XML, JSON).
Maintenant nous allons donner le fonctionnement général d’un moteur SQL sur Hadoop. De façon générale, tout comme les SGBDR MPP, les requêtes sont réceptionnées par une machine centrale (le NameNode du cluster dans certain cas, ou une machine différente qui fait office de serveur racine dans d’autre cas), ensuite celle-ci les transforme en plan d’exécution, les distribuent entre les nœuds de calcul où un processus du moteur y est continuellement en cours d’exécution. Deux cas peuvent apparaître : si le moteur s’appuie sur le HDFS alors ce processus va compiler la requête et l’exécuter directement sur le HDFS, si le moteur s’appuie sur un système de stockage différent du HDFS ou transfère les données du HDFS en format spécifique, alors la charge de calcul sera partagée entre le processus du moteur et le SGBD du système de stockage. Dans d’autres moteurs, les requêtes sont directement réceptionnées par les nœuds de calcul, transformées en plan d’exécution, compilées et traitées par ceux-ci. Nous allons voir l’implémentation concrète de tous ces principes à l’aide d’un exemple concret : Impala.

1.3. Impala : le moteur SQL sur Hadoop de Cloudera

Cloudera est un éditeur de solution logicielle, le fournisseur de la distribution commerciale d’Hadoop CDH. Cloudera est connu pour l’une de ses innovations majeures : Impala. Impala est un moteur SQL MPP sur Hadoop open source (Impala a rejoint la fondation Apache depuis 2014, mais son développement reste toujours supervisé par Cloudera). Le but d’Impala double : d’une part combiner le SQL et la performance multi-utilisateur des moteurs OLAP à la scalabilité et la flexibilité d’Hadoop, et d’autre part y ajouter les extensions de sécurité et de d’administration de Cloudera CDH. Impala permet aux utilisateurs d’écrire et d’exécuter des requêtes SQL sur des données stockées dans le HDFS ou dans HBase, exactement de la même façon que vous écrirez et exécuterez vos requêtes SQL sur des bases de données Oracle, ou SQL Server. Parce qu’il peut utiliser les métadonnées Hive, Impala est capable de lire des tables Hive et d’interpréter des scripts rédigés en HiveQL.
Comme nous l’avons dit plus haut dans l’étude des moteur SQL MPP traditionnels, l’un des plus grands challenges que pose le développement d’un moteur SQL sur Hadoop est la coordination et la synchronisation des nœuds du cluster autour des métadonnées partagées. En effet, il faut que tous les nœuds soient capables de recevoir et d’exécuter les requêtes en local, ce qui requiert qu’ils aient tous accès à la version la plus à jour des métadonnées. La majorité des moteurs résolvent ce problème en s’appuyant sur une architecture Maître/Esclave où une machine centrale héberge les métadonnées du moteur et l’utilise pour distribuer les requêtes aux nœuds de calcul. Impala quant à lui, résout ce challenge en adoptant une architecture distribué Peer-to-Peer et asymétrique, organisée autour de trois composants centraux :

• Les processus Impala, encore appelé Impalad (pour Impala Deamon), ce sont les processus qui s’exécutent en permanence sur chaque nœud de données du cluster (1 processus impalad par nœud). Ils reçoivent les requêtes SQL transmises par le biais de l’interface de commande Impala (CLI), Hue, le JDBC ou l’ODBC, lisent et écrivent les fichiers de données nécessaires, parallélisent et distribuent l’exécution des requêtes dans le cluster et transmettent les résultats intermédiaires des requêtes au Name Node. Les processus impala communiquent directement les uns des autres pour exécuter les requêtes, ce qui fait que vous pouvez soumettre vos requêtes à n’importe quel nœud de données du cluster. Chaque processus Impala est en communication constante avec le statestore pour être au courant des nœuds qui sont en bonne état et peuvent accepter des nouvelles requêtes ;
• Le Statestore vérifie l’état de fonctionnement de tous les processus Impala du cluster et leur envoie continuellement les informations de cette vérification. Le statestore est physiquement représenté par un processus appelé statestored, exécuté sur une seule machine dans le cluster. En cas de panne d’un nœud, le statestore informe tous les autres processus impala d’éviter d’envoyer les futures requêtes vers le nœud en panne ;
• Le service de catalogue, relais les mises à jour des métadonnées des instructions SQL à tous les nœuds de données du cluster. Il est physiquement représenté par un processus appelé catalogd (catalog deamon), exécuté sur une seule machine dans le cluster. Il est recommandé de faire tourner les processus catalogd et statestored sur la même machine. Le service de catalogue évite à l’utilisateur de devoir rafraîchir manuellement les métadonnées à chaque mise à jour dans les requêtes SQL ;

La figure suivante illustre l’architecture et le fonctionnement d’Impala.

Grâce à cette architecture, Impala affiche des performances plus élevées que la majorité des moteurs SQL sur Hadoop (Spark SQL, Presto de Facebook). Pour plus de détails sur Impala, veuillez vous rendre sur le site officiel de Cloudera ou sur le site Hadoop des projets en état d’incubation. Ces deux sites sont fournis à la fin de l’ouvrage. Pour le reste, la spécification SQL d’Impala reste presqu’à tout point identique à la spécification de SQL que vous connaissez déjà.

CONCLUSION DU TUTORIEL

Nous avons lu une fois sur Internet, la présentation d’un consultant qui commençait avec une citation très intéressante : « les outils ne valent que par les hommes qui les utilisent ». Dans ce tutoriel, nous vous avons montré que l’adoption à large échelle et le succès d’Hadoop ne dépendent pas des développeurs Java, mais contre tout attente des utilisateurs métiers. Ceux-ci ne se connaissent généralement pas en programmation et ont l’habitude de développer leurs applications d’exploitation de données en SQL. Facebook a été le premier à réagir à ce constat et a proposé Hive. Dans ce tutoriel, nous avons vu que Hive est une infrastructure similaire au DataWarehouse qui fournit aux utilisateurs un langage d’interrogation de données similaire au SQL : le HiveQL. Le premier problème avec HiveQL vient directement de sa nature : parce qu’il est un langage descriptif, la complexité des problèmes qu’on peut y exprimer est faible. Pour résoudre ce problème, Yahoo ! a conçu le Pig Latin, un langage de flux de données suffisamment simple à utiliser par les analystes métier. En arrière plan, ces langages transforment les requêtes des utilisateurs en job MapReduce. Avec l’évolution des besoins du Numérique, la latence du MapReduce s’est trouvé inacceptable pour les problématiques interactives, il a donc fallu penser à de nouvelles approches, et c’est ainsi que les moteurs SQL sur Hadoop sont nés. Ceux-ci s’appuient sur les principes de fonctionnement des bases de données parallèles et des SGBDR MPP qui sont utilisés pour gérer de grosses volumétries de base de données relationnelles et des DataWarehouse à l’échelle du Téraoctet. A la différence de Hive ou Pig, les moteur SQL sur Hadoop exécutent nativement le SQL sur un cluster Hadoop sans le transformer en job MapReduce. D’ailleurs, en interne, Facebook a remplacé HiveQL par l’usage de Presto, son moteur SQL MPP sur Hadoop. Tout au long du tutoriel, nous vous avons également conduit à comprendre les principes de fonctionnement des moteurs SQL MPP, de la façon dont le SQL est nativement exécuté sur Hadoop. Nous avons pris l’exemple d’Impala, le moteur SQL MPP sur Hadoop développé à l’origine par Cloudera. Comme ce livre a été écrit pour les débutants sur Hadoop, ce tutoriel devrait vous donner plus d’assurance quant à votre capacité à exploiter Hadoop. Le SQL étant une compétence commode, vous devrez normalement à ce stade être opérationnel sur l’exploitation d’un cluster Hadoop, même si vous n’avez pas une grande maîtrise de la programmation distribuée en Java, C++, ou Python. A ce stade de l’ouvrage, vous êtes opérationnels.

Si vous souhaitez allez plus en profondeur dans l’utilisation du SQL dans Hadoop, nous vous recommandons notre ouvrage : « Maîtrisez l’utilisation des technologies Hadoop »