Impossible ces temps-ci de passer une journée sans entendre parler de big data. Les consultants comme nous, nos clients, les éditeurs de logiciels, les constructeurs, les grands du web, les adeptes de la théorie du complot qui voient Big Brother à chaque coin de rue : tout le monde en parle ! Et cet article n’échappe pas à la règle.
Bien sûr, chacun met ce qu’il veut derrière ce terme, selon qu’il veut mettre en avant l’origine de big data, ses leviers, son potentiel ou ses dangers. Ici, nous nous intéresserons à la manière dont ce mouvement et les architectures informatiques s’influencent mutuellement. Les capacités de stockage et de traitement, à coût donné, sont en croissance constante, ce qui suscite de nouvelles possibilités jusqu’alors inenvisagées ; à l’inverse, ces possibilités font émerger des idées nouvelles qui stimulent la définition d’architectures toujours plus performantes. Comme nous le verrons, ce cycle continue de tourner, et nul doute que les évolutions des coûts le déplaceront dans l’avenir. Cela ne nous empêche pas de faire une photo à la date d’aujourd’hui. Nous commencerons par détailler des principes généraux (distribution, parallélisation et commoditization), avant de reconstruire couche par couche l’architecture d’un système big data tel qu’on peut le rencontrer aujourd’hui.
Dans ce beau modèle théorique où les possibilités de traitement finissent par tendre vers l’infini à coût nul, une variable nous ramène toutefois à la réalité : le débit des I/O disques et dans une moindre mesure, des réseaux. Malgré l’arrivée des technologies SSD, le débit disque croît moins vite que les autres indicateurs. Le coût de la RAM étant encore élevé par rapport à celui du stockage “dur” et non volatile, celui-ci reste nécessaire à l’absorption de l’avalanche de données nécessaire aux projets. Les projets se voient ainsi limités par le goulet d’étranglement des I/O. Stocker et traiter toutes ces données sur un ou quelques serveurs gonflés aux stéroïdes n’est pas possible à des coûts raisonnables. De toute façon, la croissance constante des volumes de données fait que cette stratégie ne serait qu’une fuite en avant. Le critère à prendre en compte n’est pas tant la capacité de stockage ou de traitement, ou le coût des infrastructures capables d’assurer ces tâches, que le coût ramené au GB de données stockées ou à la requête unitaire.
1. Distribution, parallélisation et commoditization
Pour contourner cette limitation, et se libérer d’une équation de coûts trop pessimiste par rapport aux besoins d’innovation suscités par le big data, il a fallu changer radicalement les architectures. Celles mises en oeuvre dans tous les projets big data répondent par trois stratégies complémentaires : distribution des données, parallélisation des traitements et commoditization des infrastructures. Ces stratégies sont directement inspirées des travaux des géants du web (www.geantsduweb.com), Google, Facebook et autres Yahoo!, qui ont été les premiers à faire face aux limites mentionnées plus haut.
Distribution des données : pour fixer les idées, disons que l’on parlera de big data à partir d’un ordre de grandeur de 10 To de données en ligne. Si un serveur ou un SAN n’est pas capable, de manière économique, d’héberger le volume nécessaire aux traitements, alors il faut distribuer ce volume sur plusieurs serveurs. Cette distribution est au minimum horizontale, c’est-à-dire que les ensembles d’objets à stocker sont regroupés en paquets -- des shards -- répartis sur les nœuds du cluster. Le critère de répartition est souvent le résultat d’une opération de hachage sur les données. À cette stratégie de distribution, le sharding, s’ajoute parfois un partitionnement des données individuelles ; cette fois, les objets eux-mêmes peuvent être découpés et répartis sur plusieurs nœuds (c’est le cas, par exemple, des fichiers HDFS, le système de fichiers d’Hadoop). Pour récupérer un objet complet, il faut alors identifier les nœuds qui hébergent chacune de ses parties et les recomposer.
Parallélisation des traitements : les besoins tendent aussi vers une réponse de plus en plus rapide aux problématiques métiers. Au vu des volumes en jeu, les traitements séquentiels ont des temps d’exécution qui sont prohibitifs : la parallélisation est nécessaire. Notons déjà que ce point entretient un rapport étroit avec le précédent : les goulets d’étranglement se situant au niveau des I/O des disques et des équipements réseau, une stratégie de parallélisation efficace doit forcément tenir compte de l’emplacement des données manipulées. C’est la notion de colocalisation (nous y reviendrons juste après).
Commoditization des infrastructures : derrière cet anglicisme se trouve l’idée qu’il est plus économique, à terme, de remplir des armoires avec de nombreux serveurs de série (commodity hardware), qu’avec quelques serveurs hauts de gamme. Ces serveurs “basiques” assument le double rôle de distribution du stockage et du traitement que nous venons d’évoquer. Encore une fois, il faut considérer le coût ramené au GB de données. Cette stratégie confère aux clusters ainsi constitués une propriété d’élasticité : on peut ajouter des noeuds de stockage et de traitement au fur et à mesure de l’évolution des besoins, progressivement, sans faire de gros investissements par à-coups. On peut aussi en enlever ! C’est ainsi que les géants du web ont constitué, petit à petit, des clusters de plus en plus gros pour faire face à leurs traitements de plus en plus gourmands.
1.2 Nouvelles stratégies, nouveaux défis
Si ces stratégies s’avèrent payantes tout en semblant reposer sur des principes simples, il ne faut pas négliger non plus les défis qu’elles suscitent.
Tout d’abord, au niveau des plateformes techniques, il s’agit ni plus ni moins que de mettre en place des systèmes distribués, avec leur cortège de problématiques : cohérence et équilibrage des données distribuées, tolérance aux pannes, distribution multi-datacenters, cohabitation de traitements transactionnels et analytiques, etc. En particulier, les plateformes distribuées embarquent nécessairement des mécanismes de résilience face à la perte d’un noeud, d’autant plus lorsque l’infrastructure est de type commodity, plus encline à la panne.
La résilience pour le stockage implique nécessairement la duplication de la donnée entre deux ou plusieurs noeuds. Cette duplication offre aussi un avantage non négligeable pour l’exécution : elle offre une grande souplesse dans l’affectation des ressources de calcul aux nœuds, en appliquant le principe de colocalisation évoqué précédemment. Ce principe veut que, plutôt que de rapatrier un shard à traiter sur un nœud de calcul distant, c’est le calcul qui est envoyé vers le nœud qui héberge une des répliques du shard, par exemple le noeud qui est le moins sollicité à ce moment. Les I/O réseau sont minimisés lorsque c’est possible. La duplication des shards a ainsi un double avantage, pour un inconvénient mineur étant donné le coût des disques : une empreinte de stockage multipliée par le facteur de réplication. Il faut toutefois en tenir compte au moment de dimensionner la plateforme !
Nous pouvons résumer par un schéma le couplage entre le stockage et les traitements (Figure 1).
Deuxième défi, les changements d’architecture amenés par ces nouveaux paradigmes interviennent au niveau le plus bas. Il a donc fallu revoir toutes les couches qui s’appuient dessus, et pour ainsi dire réinventer la roue à tous les niveaux. C’est ce que nous allons voir plus en détail maintenant, en visitant l’une après l’autre les couches d’architecture qui constituent un système “big data” générique (nous ne ferons qu’à la fin la distinction entre les systèmes transactionnels -- type NoSQL -- et analytiques -- type Hadoop). Nous commencerons par le bas, pour empiler les couches.
2. L’infrastructure
Le stockage et les traitements s’appuient évidemment sur une infrastructure qui en fournit les ressources : dans le cas le plus général, ce sont des disques, des cœurs de calcul, de la RAM et du réseau.
Nous avons vu que ces ressources étaient nécessairement distribuées, du fait des volumes en jeu (10 To et au-delà) et de la lourdeur des traitements qui opèrent dessus. L’infrastructure d’un système big data est une infrastructure de cluster. Elle a besoin d’une interconnectivité réseau entre les noeuds du cluster. La topologie réseau n’est pas anodine, car ces mêmes volumes de données sont échangés de manière importante entre les nœuds. Les liens doivent être suffisants pour ne pas être un goulet, et la plateforme doit idéalement tenir compte de la topologie pour optimiser le placement des données et des traitements en minimisant les échanges (par exemple en favorisant les transferts entre nœuds d’un même rack).
Enfin, dans le cas d’un hébergement sur machines physiques, une couche de virtualisation peut être nécessaire. La virtualisation va en quelque sorte à l’encontre de l’approche mentionnée plus haut, car elle permet la consolidation des ressources de stockage ou de calcul, et non leur extension. La virtualisation, lorsqu’elle existe, ne répond pas tant à un besoin d’utilisation optimale des ressources que d’élasticité et de facilité de provisioning des machines.
3. Stockage vs modélisation des données
L’observation la plus évidente que l’on peut faire est la séparation nette entre le stockage et la modélisation des données : cela constitue une rupture par rapport aux systèmes relationnels classiques, où la base de données embarque son schéma.
En effet, le besoin de traiter des gros volumes de données, hétérogènes (voire non structurés) et variables dans le temps, milite pour une modélisation lâche ou même inexistante. Le cas extrême se retrouve du côté des bases de données NoSQL en clefs/valeurs, qui ne sont ni plus ni moins que des tables de hachage distribuées. Leur API est réduite au minimum : get(key) et put(key, val) ; les valeurs sont pour la base des BLOBs opaques et charge à l’application utilisatrice d’interpréter leur contenu. D’autres bases de données, orientées document (où les structures de données sont mémorisées d’un seul tenant, comme des objets complets) ou en familles de colonnes (où les attributs de chaque ligne sont stockés ensemble par groupe, les familles), proposent un stockage un peu plus structuré quoique encore bien plus lâche que les modèles relationnels. Les solutions de traitement en masse, comme Hadoop, n’échappent pas à cette règle puisqu’elles ne “voient” que des fichiers, i.e. des flux indifférenciés d’octets.
Toutes ces situations ont une caractéristique commune : quand il existe, c’est l’application qui “fait” le modèle, le maintient et assure la compatibilité ascendante des données qu’elle rencontre. Pour faire face aux évolutions inévitables de ce modèle dans le temps, plusieurs stratégies sont possibles ; selon les frameworks utilisés et les patterns d’accès à la donnée, telle ou telle stratégie sera plus adaptée.
3.1 Évolutivité des modèles de données
Le plus simple est une reprise d’historique des données in-situ lors des migrations, comme aujourd’hui avec un SGBDR -- mais avec des volumes bien plus importants bien sûr.
Une deuxième stratégie, appelée schema on read, est l’interprétation à la lecture du modèle de la donnée lue. C’est la stratégie la plus courante dans les environnements de type Hadoop, car elle se prête aussi bien aux données structurées et non structurées, et permet à plusieurs applications ou requêtes d’interpréter chacune une même donnée à sa manière. Elle induit bien sûr un surcoût répété à la lecture, qui peut être pénalisant sur des accès massifs.
La troisième stratégie est plus adaptée aux systèmes transactionnels, soumis à des lectures unitaires fréquentes qui doivent rester performantes. Elle s’appuie sur la stratégie schema on read pour transformer la donnée de l’ancien modèle vers le nouveau, et la réécrire immédiatement afin de faciliter les lectures suivantes. C’est une sorte de reprise d’historique opportuniste...
3.2 Modélisation et distribution
Nous avons présenté le découplage stockage / modélisation comme une nécessité liée à la forme et à l’évolution des données elles-mêmes. Toutefois, on peut justifier ce choix d’architecture par d’autres critères, plus techniques : la contrainte de distribuer et dupliquer les données conduit de toute façon à abandonner la modélisation relationnelle, trop contraignante.
Certes, les SGBDR modernes savent faire du sharding, mais au prix de certaines contorsions pour respecter le modèle relationnel tout en gardant des performances acceptables. Ainsi, le respect des contraintes de clefs étrangères, pour être efficace, suppose que des enregistrements corrélés soient stockés ensemble ce qui, en jargon de modélisation NoSQL, n’est pas très loin d’un modèle orienté document... Ou bien certaines tables, comme les tables de référence, devront être dupliquées pour satisfaire les relations sur tous les nœuds séparés.
En revanche, des modèles de stockage en clefs/valeurs ou documents se prêtent très bien à la distribution, car les contraintes de clefs étrangères n’existent plus. C’est bien sûr à l’application d’assurer les règles d’intégrité. Quant aux systèmes à base de fichiers, comme HDFS, la question ne se pose pas : ce sont des flux d’octets que l’on peut couper où l’on veut, pour recoller les morceaux à la lecture.
4.Traitement des données
Nous venons de voir que si la distribution physique des données est assurée par la plateforme de stockage, et donc transparente pour l’application, ce n’est pas le cas de sa structure.
Lorsque la donnée stockée est exploitable directement (par exemple elle est déjà assez structurée, et le traitement consiste en un calcul de statistiques simples comme un comptage ou une moyenne), le code du traitement peut directement opérer dessus. Selon les plateformes, l’implémentation devra tenir compte ou non d’une exécution parallèle du traitement par le framework. C’est le cas des plateformes s’appuyant sur un framework MapReduce, par exemple, qui imposent une structure particulière aux programmes qui l’utilisent.
Cependant, dans de nombreux cas, la donnée issue du système de stockage n’est pas exploitable en l’état par le coeur du traitement. Il faut d’abord la faire passer dans un pipeline d’interprétation, pour lui donner une forme consommable par l’algorithme. Ce pipeline peut bien sûr être implémenté par le programme lui-même, ou faire appel à des librairies pré-existantes que le programme se contentera de coordonner.
Avant de citer quelques unités de traitement possibles pour un tel pipeline, notons que ce dernier peut s’exécuter de différentes manières : en tâche de fond sur de gros volumes (mode batch), sur des volumes faibles ou avec des temps de réponse garantis faibles (mode interactif), ou au fur et à mesure de l’arrivée des données unitaires dans le système (mode événementiel).
4.1 Les étapes du pipeline
Une première étape quasiment obligatoire, avec la variété des sources de données à laquelle on est confrontée, est la préparation des données. Il faut en général les nettoyer, les enrichir (avec des données de référence ou des valeurs par défaut pour les portions manquantes), afin qu’elles atteignent une qualité suffisante pour être réellement traitées.
L’indexation des données est une étape nécessaire lorsqu’un besoin de recherche dans les données est identifié. Il n’est en effet pas question de faire une recherche brutale sur des volumes aussi importants, pour retrouver une donnée précise. Les solutions de type NoSQL, qui stockent des objets, offrent des index de type clef primaire. On observe cependant que les besoins plus élaborés (index secondaires, recherches plein-texte) sont en général confiés à des moteurs de recherche tiers (type Solr, Lucene) intégrés avec le stockage, et qui opèrent directement sur leurs données.
Pour les données non ou faiblement structurées (texte, images, vidéo...), il est nécessaire de faire une extraction de métadonnées, ou des traitements spécifiques au type de document. À titre d’exemple, pour du texte on trouvera une extraction de mots, des statistiques sur les paragraphes, voire du traitement élaboré de langage naturel (NLP --Natural Language Processing) et une interprétation sémantique. Pour des images, on peut envisager d’en calculer la taille, de faire de l’OCR, de la détection de visages ou de logos... Cette première étape du pipeline permet de doter la donnée brute d’une structure exploitable par les étapes suivantes. Elle peut aussi alimenter un moteur de recherche.
Dans de nombreux cas, on est amené à appliquer des méthodes statistiques plus poussées que de simples sommes, afin d’extraire plus d’information ou de construire des modèles statistiques. C’est le domaine du machine learning, qui n’est ni plus ni moins que l’incarnation moderne et (partiellement) automatisée des techniques traditionnelles de data mining. Il sert trois problématiques : décrire, expliquer et prédire. Les algorithmes font appel à des combinaisons de techniques statistiques, d’intelligence artificielle, et bien sûr au flair de l’analyste de données (le fameux data scientist) qui va choisir les meilleures stratégies en fonction du contexte.
On peut voir les traitements des graphes comme un cas particulier du machine learning, mais il a aussi ses caractéristiques propres. De nombreux problèmes peuvent être formulés avec une représentation en graphes. C’est une représentation compacte d’objets et de leurs relations, qui offre l’avantage d’être outillée par une théorie mature. La panoplie offerte par cette théorie est vaste, et couvre un spectre d’analyse très large : analyse descriptive globale sur les propriétés du graphe, calcul d’influence des sommets au sein du réseau, calcul de chemins, etc. Les défis, avec les graphes, viennent des arcs dans les scénarios de parcours. Lorsque le graphe doit être distribué sur plusieurs noeuds de stockage, l’efficacité du partitionnement dépend directement de la répartition de ces arcs.
Voici la vue d’ensemble d’un pipeline complet (Figure 2).
5. Requêtage et visualisation
Au-dessus des étages de traitement et de préparation, nous trouvons l’exploitation des données proprement dites. Les patterns de mise à disposition des données n’ont pas tellement changé : préparation de dashboards statiques, mise à disposition d’outils d’exploration des données, « push » de mises à jour sur des données qui changent fréquemment, ou tout simplement ouverture d’un infocentre à des requêtes ad hoc.
Des données, brutes ou structurées par le biais d’une étape précédente, peuvent tout d’abord être requêtées. Il s’agit ici de transposer, sur une plateforme big data, les requêtes que l’on fait depuis longtemps dans le monde SGBDR classique : recherches, accès par clef primaire, enrichissement (jointures), statistiques descriptives simples (comptage et somme typiquement) avec regroupements. L’apport de la plateforme big data consiste à brasser des quantités de données inaccessibles aux SGBDR. Entre les systèmes NoSQL, essentiellement transactionnels, et les systèmes BI comme Hadoop et Hive, il existe des différences considérables dans les types de requêtes que l’on peut faire de manière efficace -- même si une convergence est à l’oeuvre -- nous y reviendrons en fin d’article.
Le volume des données étant en général énorme, en faire une représentation fidèle et synthétique, qui sache présenter et expliquer, est un défi ; c’est le domaine de la visualisation. Les tableaux et les camemberts de papa ont encore de beaux jours devant eux, mais ils ne suffisent plus, car ils sont purement descriptifs (ils n’expliquent rien des traitements élaborés qui ont conduit à la production des données), ne permettent pas de traduire la richesse des analyses multi-variables ou des données connectées (graphes). Et tout simplement, ils ne sont pas assez “jolis” pour des utilisateurs de plus en plus exigeants. Les éditeurs, et les contributeurs open source, proposent des catalogues de représentations très riches, statiques ou interactives. À titre d’exemple, on pourra consulter la galerie du site D31.
Même au niveau restreint de la couche de visualisation, les architectures évoluent. Si la préparation de reporting prédigérés peut être vue comme une étape de traitement parmi d’autres sur la plateforme big data, les visualisations à base d’exploration et de requêtage ad hoc se heurtent aux problèmes de performance que les architectures batch ne savent pas toujours contourner. Cette limitation est levée par l’emploi d’architectures mixtes batch/TP, qui sont en train d’émerger et d’enrichir considérablement l’offre logicielle des plateformes.
6. Intégration au SI...
... ou à l’extérieur du SI ! Pas de big data sans données, c’est une évidence : il faut bien amener ces données sur la plateforme, avant de l’utiliser. Et dans l’autre sens, on veut pouvoir exporter les données traitées vers des fichiers ou d’autres systèmes, par exemple stocker des agrégats de haut niveau dans des entrepôts relationnels classiques. Pour tout cela, des connecteurs sont nécessaires.
On trouve dans cette catégorie des outils d’import/export depuis/vers les bases de données du SI ou des fichiers. Les API, et outils en ligne de commandes, sont aussi de cette catégorie, car ils vont permettre à des applications externes d’interroger les données stockées sur la plateforme (voire de lancer des traitements).
7. Management de cluster
On retrouve dans cette couche les outils habituels de gestion d’infrastructure, adaptés aux composants opérationnels de la plateforme big data. Les outils et composants sont assez inégaux de ce point de vue ; nous allons illustrer ce point avec deux exemples.
Tout d’abord la supervision : les solutions offrent en général peu de métriques de haut niveau permettant de comprendre l’utilisation du cluster, d’un point de vue de la charge de travail. Le dimensionnement, le capacity planning, sont donc encore assez artisanaux... Ou bien ces métriques existent, mais ne sont pas centralisées facilement (c’est le cas d’Hadoop).
La sécurité, elle aussi, est souvent laissée de côté. Elle constitue pourtant un enjeu à elle seule, car l’enjeu de big data étant l’exploitation massive de toutes les données à disposition (en caricaturant), la protection de ces données est cruciale. Là encore, Hadoop dispose d’un système de sécurité encore embryonnaire, mais que les prochaines versions d’HDFS vont compléter -- en introduisant également des politiques de gestion de cycle de vie de la donnée.
8. Synthèse : l’architecture d’un système big data
Nous sommes maintenant en mesure de dresser le portrait général d’un système big data, en reprenant les couches que nous avons vues, sous forme d’un fond de carte (Figure 3).
Bien entendu, tous les systèmes n’implémentent pas toutes les couches ; en particulier les systèmes transactionnels et analytiques se distinguent par leurs couches hautes. Du moins était-ce le cas jusqu’à récemment... la situation devient plus floue.
Nous avons déjà mentionné la convergence entre les systèmes doués pour le transactionnel, et ceux plus portés sur l’analytique. Elle se manifeste par le fait que les deux classes d’architecture, différentes dans leur conception, se voient augmentées de capacités empiétant sur l’autre royaume. Du côté des bases NoSQL, de plus en plus se dotent de frameworks MapReduce (MongoDB, Riak) ou proposent une intégration avec Hadoop en se substituant au système de fichiers HDFS (l’édition DataStax de Cassandra, avec CFS). À l’inverse, côté Hadoop, une compétition féroce est lancée pour savoir qui proposera le framework de requêtage interactif de demain (Impala chez Cloudera, Hive Stinger chez Hortonworks, Pivotal HD chez VMWare, sans compter des améliorations de MapReduce comme Tez ou Spark).
Cependant, il n’y aura pas d’architecture technique unique estampillée “big data”, au-delà de la description fonctionnelle que nous venons de faire. Chaque implémentation étend son périmètre, mais au prix de compromis (sur le stockage, la duplication de données ou les performances absolues). Le choix de la plateforme reste très influencé par le jeu de requêtes auquel les données vont être soumises, et comme toujours, le système idéal pour votre cas d’utilisation sera un hybride intelligent du meilleur de chaque plateforme.
Pour conclure, je tiens à remercier une bonne quinzaine de mes collègues qui participent aux ateliers de réflexion et de rédaction pour le livre blanc sur le big data que je coordonne actuellement.