Création d'un socle big data ( partie 4 )

Pour cette dernière partie, je vais vous parler de la v360, pas au niveau de l'interface, là, vous trouverez toute la littérature utile sur la toile, mais d'une implémentation possible pour gérer les données.

Elle s'appuie sur une stack Hadoop ( Cloudera ) et elle repose sur les services Hive, HBase, Lily Indexer et Solr. L'objectif est de proposer une exposition simultanée des données en mode transactionnel ( Hbase ) et en mode recherche ( Solr ).

Pour ce faire, les données de la table Hbase sont mises à jour en mode map-reduce à partir des données raffinées du data lake. Puis, Lily Indexer s'occupe de la réplication des données entre la table Hbase et la collection Solr correspondante.

Maintenant, rentrons un peu dans la technique. Le table HBase est définie de cette manière :

CREATE TABLE IF NOT EXISTS table_name(row_key type, col1 type1, …, coln typen)
STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf1:col1,…,cf1:coln")
TBLPROPERTIES ("hbase.table.name" = "namespace_name:table_name");

Cette définition permet à Hive d'utiliser la table, cette dernière étant déclarée dans le métastore Hive.

Dans le socle, il existe toute une gestion du cycle de vie de la table Hbase ( création, remplacement, suppression ... et même un undo du remplacement ! ). De plus, cette table peut être indexée, c'est à dire on peut lui associer une collection Solr.

Cette dernière peut être définie ainsi:

solrctl instancedir --generate /root_socle/environnement/socle/solr/collection_name
cp directory_conf_usage/collection_name_schema.xml  /root_socle/environnement/socle/solr/collection_name
solrctl instancedir --create collection_name /root_socle/environnement/socle/solr/collection_name
solrctl collection --create collection_name -s num_shard -r num_replica

La partie la plus délicate consiste à configurer le service Lily indexer. Le point de départ est ce lien très général, mais je vous conseille de vous faire aider par le support Cloudera pour cette partie si possible. Sinon, vous pouvez vous appuyer sur ce tutoriel très complet et vous y réussirez petit à petit en mode essai/erreur ...

Pour information, nous avons implémenté Kerberos et Sentry et cela marche correctement sur nos clusters de dev. Reste à le valider en pré-production, puis en production.

Une fois cette étape franchie, il vous reste à définir la configuration de la réplication ( deux fichiers ) et sa déclaration ( commande hbase_indexer ). Le mode d'emploi se trouve ici.

Une illustration du fichier morphline pour le type double ( ajout de code Java ) :

DDL:

CREATE TABLE IF NOT EXISTS tab_double(id string, col1 double)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf1:col1")

TBLPROPERTIES ("hbase.table.name" = ":tab_double"); 

Morphline :

{

extractHBaseCells {

mappings : [

{

inputColumn : "cf1:id"

outputField : "id"

type : "string"

source : value

}

{

inputColumn : "cf1:col1"

outputField : "col1"

type : "string"

source : value

}

]

}

}

{

java {

imports : "import java.util.*;"

code: """

String stringAmount = (String) record.getFirstValue("col1");

logger.info("stringAmount:" + stringAmount);

Double dbl = Double.parseDouble(stringAmount);

record.replaceValues("col1",dbl);

return child.process(record);

"""

}

}

Schema.xml :

fields
   field name="id" type="string" indexed="true" stored="true" 
                   required="true" multiValued="false"
   field name="col1" type="double" indexed="true" stored="true"
   field name="_version_" type="long" indexed="true" stored="true"
fields

Puis, on crée l'indexation :

$ hbase-indexer add-indexer \
--name table_name_indexer \
--indexer-conf root_socle/socle/solr/collection_name/
               morphline-collection_name-mapper.xml \
--connection-param solr.zk=zkhost1:2181,zkhost2:2181,
               zkhost3:2181/solr \
--connection-param solr.collection=collection_name \
--zookeeper zkhost1:2181,zkhost2:2181,zkhost3:2181

La réplication dispose d'un mode debug fort utile lors du développement, le mode par défaut étant peu disert.

Création d'un socle big data ( partie 3 )

Après avoir présenté le socle en terme d'architecture technique, détaillons un premier usage, qui ne sera pas une v360, premier usage habituel pour un data lake ( au prochain épisode ! ), mais un outil BI couplé à Hadoop.

L'idée est ambitieuse, elle consiste à vouloir remplacer progressivement un infocentre ( Oracle, puis DB2 ) par une solution Hadoop pour effectuer des requêtes analytiques plus rapidement et même de pouvoir exécuter de nouvelles requêtes qui ne se terminent pas habituellement.

Pour ce faire, le choix s'est porté sur Impala, les données étant stockées sous HDFS soit en format texte, soit en format Parquet pour optimiser le stockage et les performances. On n'a pas introduit Kudu pour de multiples raisons: manque de maturité, Kerberos à l'époque pas implémenté, PRA non disponible, ...

Impala est un service dédié à l'analytique, de nombreuses présentations sont disponibles sur slideshare pour bien cerner la technologie, par exemple : Impala internals.

La recette étant bien avancée, l'objectif est en passe d'être atteint, il reste encore sur le premier lot à réécrire quelques requêtes adaptées à l'optimiseur Oracle, mais pas à l'optimiseur Impala. Mais le nombre est restreint comparé au nombre de requêtes migrées.

Une première remarque, n'oubliez pas de calculer les statistiques sur les tables suite à une mise à jour importante, il n'existe pas encore une fonctionnalité permettant le recalcul automatique des statistiques si nécessaire.

Impala est un système effectuant des scans en parallèle sur plusieurs noeuds, la scalabilité étant assurée par l'ajout de noeuds si les performances se dégradent. De plus, il nécessite une quantité de RAM non négligeable ( 256 GB RAM par noeud ).

Par conséquent, le tuner ne dispose pas d'index pour optimiser ses requêtes. Il doit donc porter une attention toute particulière au choix de ses jointures. Heureusement, Cloudera dispose d'une grammaire riche dans ce domaine et vous vous apercevrez vite que parfois, il faudra changer le type de jointure pour éviter des temps de réponse trop élevés.

Pour des tables de grande volumétrie, il vous faudra aussi penser à partitionner vos tables pour éviter des jointures trop importantes.

Il n'existe pas encore de livres sur le tuning Impala de la qualité de ceux disponibles sur Oracle ( J.Lewis, Antognini, ... ). La seule documentation est celle offerte par Cloudera, heureusement de qualité. Le point de départ est ici.

Pour finir, un dernier tip: il existe bien des hints pour Impala et le hint STRAIGHT_JOIN vous permettra de reprendre la main sur les jointures.

Création d'un socle big data ( partie 2 )

Le socle applicatif est écrit sur la base de trois langages ( bash linux, Python, Java ).

Le bash articule le coeur du socle et de nombreuses parties, en particulier la gestion des structures de données, sont développées en shell. Pourquoi ? Il permet d'utiliser les binaires de la stack Cloudera ( beeline, hbase shell, solrctl, hbase-indexer, ... ) et de cette manière, on bénéficie de toutes les fonctionnalités pas toujours disponibles dans des API de plus haut niveau.

Le bash est bien entendu un langage difficile à appréhender, mais on peut désormais l'écrire de manière maintenable et modulable, avec pour notre part l'utilisation de "contextes" qui sont en fait des librairies réutilisables par le socle. De plus, si pour certaines opérations comme par exemple l'analyse de fichiers xml, le bash n'est pas l'outil idoine, loin de là, on fait alors appel à Python, un interpréteur très en voque dans le big data, qui a le mérite d'être plus souple qu'un langage compilé comme Java.

J'apprécie le bash car c'est un outil sans équivalent en terme d'utilitaires. Un exemple: l'utilisation de taskset pour attribuer une tâche à un CPU.

On a ajouté du Python pour des besoins ponctuels, mais aussi car c'est l'outil favori des data scientists, même si avec l'avènement de la plate-forme spark et de scala, il sera intéressant de voir si le machine/deep learning ne migrera pas dans une certaine mesure vers ce type d'infrastructure où tout est optimisé pour la parallélisation des traitements et des données ( rdd ) dans des environnements distribués.

Initialement, on a développé le socle avec des outils tels putty, notepad++, ... bref des outils peu appréciés par certains développeurs peu habitués à linux. C'est pourquoi on propose aussi de développer sur Eclipse et donc de baser le point de départ du mécanisme d'intégration continue sur cet outil. Pour faciliter la tâche des futurs développeurs du socle, on a aussi écrit quelques outils en Python histoire de nous familiariser davantage avec ce langage.

Enfin, on utilise Java car une partie du data loader repose sur un composant Java précédemment écrit chez un autre client. De plus, il faut savoir que Java est un langage incontournable chez les grands comptes français.

En tant qu'ex-consultant Oracle, je suis sensibilisé par la gestion des structures de données et le socle contient donc un outil qui permet d'assurer le déploiement des mêmes structures de données sur tous les environnements de l'infrastructure. Cela paraît anodin comme ça, mais je peux vous assurer que c'est très rare dans les faits ...

De plus, autant à terme dans le big data 2.0, vous aurez des data loader, des outils de streaming, des outils de crunching, ... en mode graphique à l'image des anciens ETL, autant la gestion des structures de données sera probablement comme pour les bases relationnelles la grande oubliée de l'histoire car peu génératrice de business ...

On gère déjà pas mal de fonctionnalités ( droits HDFS, ACL, Sentry, partitionnement, format de fichiers, ... ) et la manière dont a été conçu cet outil permettra d'ajouter d'autres fonctionnalités de manière souple, par exemple des éléments de performance si nécessaire.

Le data loader purement technique est enrichi d'une couche de tables au sens Hive compatible Impala qui permet d'avoir des référentiels de données d'une part et une historisation des données entrantes d'autre part. Il fera à terme, il faut l'espérer, le bonheur des data scientists pour leur exploration des données en mode brut.

Création d'un socle big data ( partie 1 )

Depuis plus d'un an, je pilote un socle applicatif autour de la stack Cloudera afin de la rendre disponible à des usages ( = applicatif métier utilisant les données centralisées dans un data lake ).

La stack Cloudera est un package de multiples services ( hive, impala, solr, spark, ... ) administrable et déployable via un outil de management, cloudera manager, bien utile pour les équipes d'infrastructure, mais en terme applicatif, tout reste à  faire !

L'idée générale est de créer des patterns ( = services ), des fonctions réutilisables pour l'ensemble des usages leur permettant de pouvoir traiter de la data par de multiples canaux divers et variés.

Bien entendu, ces patterns doivent être déployables sur des infrastructures multi-noeuds, plus précisément un environnement logique composé d'un ensemble de patterns est susceptible d'être installé sur plusieurs environnements multi-noeuds.

Pour ce faire, on a écrit un outil de génération d'environnement en bash s'appuyant sur maven et on utilise un outil d'intégration continu propre au client pour déployer de manière automatique un environnement logique sur toutes les plates-formes disponibles ( qqes centaines de cpu, qqes TB RAM et qqes centaines de TB disque pour vous donner une idée de ce qu'est une infra big data ).

Tout cela pour vous dire que tout est à créer dans le big data 1.0 en mode non cloud !

Bon, revenons à nos patterns. On peut distinguer plusieurs types de patterns, ceux qui permettent de manager le socle ( outil de déploiement, outil de gestion des structures de données, ... ), ceux qui perment d'alimenter le data lake ( loader en mode batch, streaming de flux de données, reprise de données via sqoop ... ), ceux qui permettent d'épurer la donnée loin d'être très propre ( crunching via un outil comme Talend ) et ceux qui offrent  enfin ( ! ) une plus-value pour les usages ( batch de traitement des données en map/reduce, traitement BI via des outils comme Impala et Kudu, V360, data science pour les plus avancés, ... ).

La genèse d'un tel système est tout sauf une sinécure. Après une phase d'architecture pour bien poser l'infrastructure et les besoins des usages, on peut alors d'une part commander l'infrastructure, l'installer et la rendre opérationnelle ( plusieurs mois ) et d'autre part sur des machines mono-noeuds construire les premiers patterns pour alimenter le data lake ( structures de données, chargement des données ).

Une fois l'infra posée, on peut réfléchir à l'outil de construction et de déploiement des patterns tout en commençant à nettoyer et à enrichir les données présentes dans le data lake. A ce moment-là, les usages commencent à avoir accès aux données via JDBC pour effectuer des requêtes en mode SQL ( hive, impala, ... )  et commencer à aller plus loin via des outils de dataviz comme Tableau ou SAS.