Le SQL dans Hadoop - Partie 2 : les moteurs natifs SQL sur Hadoop

Avec la baisse des coûts des 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 ;
• partitionner les données de la base de données ;
• 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, par des processus indépendants appliqués aux différents nœuds contenant les partitions de la base de données. Comme chaque nœud contient une partie de la base de données, il s'ensuit 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 ou les nœuds considérés, une même requête pouvant toucher différentes partitions.

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 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 à l'identique entre tous les nœuds d'un cluster Shared-Nothing et les données sont distribuées sur les différentes partitions à travers ces nœuds, de sorte que chaque nœud héberge une partie de la base de données. De plus, toutes les communications dans le cluster se font 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, un même client n'étant stocké que dans une seule partition, un seul nœud du cluster. 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 rassemblement 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, le stockage de la base est centralisé et chaque nœud possède en mémoire une copie de la base. En revanche, dans une architecture Shared-Nothing, chaque nœud possède sa propre mémoire et son propre stockage, par conséquent la base de données doit être partitionnée et distribuée pour chacun des nœuds du cluster.

Dans cette dernière architecture, chaque nœud ne peut donc 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. En effet, l'architecture Shared-Nothing, impose que les tâches ne soient pas dépendantes de plusieurs nœuds en vertu de l'exigence de parallélisme du traitement des requêtes. Autrement dit, chaque partie de la requête s'exécute sur un nœud sans jamais interagir avec un processus concurrent de la même requête.

Le partitionnement de la base de données doit donc se faire de façon à ce que les partitions soient les plus indépendantes possible. À quelques exceptions près, 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 originale. Celui-ci consiste à diviser une table par intervalles de lignes, sans chevauchement et sans « trou ». Il se différencie du partitionnement vertical qui consiste à diviser une table par colonne, la clé servant de référence commune pour retrouver les données d'une même ligne. 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.

Une fois obtenu les l partitions de lignes d'une même table, la question se pose de l'accès aux différents nœuds. Autrement dit, comment le système répartit-il les partitions sur l'ensemble des nœuds du cluster ? Pour résoudre ce problème, il existe principalement trois méthodes de répartition des données : 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, ou une clé alternative non nulle, de la table qui est utilisée comme clé de partitionnement. Chaque nœud reçoit alors un intervalle de valeurs de la clé. Cette méthode possède l'avantage que la clé agit comme un index, ce qu'elle est naturellement. Mais il faut s'assurer que la distribution des valeurs de la clé assure une répartition relativement équilibrée des lignes entre les nœuds et le reste au fil du temps, et que les valeurs de cette clé soient immuables.
• 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 répartit les lignes de la table à raison d'une ligne par nœud de manière circulaire. Le premier nœud reçoit la ligne 1, le second la 2e, etc., et une fois atteint le dernier nœud, on cycle la distribution des lignes en revenant au premier nœud, tout cela de manière indépendante de la valeur de la clé. On a alors l'assurance que les lignes seront aussi distribuées que possible. L'inconvénient est qu'on ne sait plus sur quel nœud une ligne particulière est située, ce qui oblige à scruter tous les nœuds en parallèle pour la plupart des requêtes.
• 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é afin d'obtenir une distribution pseudo-aléatoire des lignes sur les nœuds et par conséquent des partitions relativement homogènes en termes de cardinalité. Comme dans l'affectation par intervalle, le hachage assure également que les lignes qui ont une valeur de clé identique soient dans la même partition et donc affectées au même nœud (colocalisation des données de même valeur de clé). Ainsi, les jointures sont plus simples à réaliser et il n'y aura pas besoin de déplacer les 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 trois 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 deux 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écutive, 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.

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 utilisateurs.  En réalité, le SQL fonctionne également comme une abstraction, elle permet à l'utilisateur de décrire le QUOI  sans décrire le COMMENT, c'est d'ailleurs le but de tout langage de requête par nature. 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, après algébrisation et optimisation, ce qui conduit à livrer 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 appliqué à un SGBDR parallèle en particulier, le processus général est le même, mais le schéma d'exécution diffère. En effet, 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 partition de la base de données. Ce sont donc les mêmes instructions (la même requête SQL) qui sont exécutées simultanément sur chacun des 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 compilent et l'exécutent. 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 restriction (opération consistant à retrouver certaines lignes d'une table) en recherchant les lignes pour lesquelles l'identifiant du produit vaut 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. Étant donné que cette requête filtre sur une valeur de 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. La performance est alors meilleure 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 le nœud de référence d'Hadoop, 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 between 10001 AND 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 trois nœuds de calcul). Les trois 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 concaténer pour obtenir la table finale de résultat.

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 6 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 comme illustré 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.

Voilà, 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.

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 coûteux financièrement qu'Hadoop. Coûteux financièrement en termes de logiciel, de matériel, d'administration, et même en termes 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 trois 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. Rares 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'autres 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.

Cloudera est un éditeur de solutions logicielles, 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 est double : d'une part combiner le SQL et la performance multiutilisateur des moteurs OLAP(1) à la scalabilité et la flexibilité d'Hadoop, et d'autre part y ajouter les extensions de sécurité et 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 écririez et exécuteriez 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 moteurs 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ée 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 (un 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 avec les 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 bon état et qui peuvent accepter de 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 relaie 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. Pour le reste, la spécification SQL d'Impala reste presque à tout point identique à la spécification du SQL que vous connaissez déjà.