\chapter{Gestion de données large-échelle et systèmes~distribués}
\label{ch:gestion}

Ce chapitre traite de la gestion de données à large-échelle. Nous commençons par décrire les limites des systèmes traditionnels de gestion de données. Nous décrivons ensuite l'architecture Lambda qui a été proposée ces dernières années pour le traitement de grandes masses de données. Nous présentons ensuite successivement les trois couches de l'architecture Lambda : batch, service et vitesse.  

\section{Les limites des systèmes\\traditionnels de gestion de données}\label{s:besoins_limites} 

Nous débutons cette section par une description des besoins liés au traitement de grandes masses de données. Nous expliquons ensuite les limites des systèmes traditionnels de gestion de données. 

%besoins liés aux traitements de grandes masses de données. Nous présentons présente la mise en oeuvre des systèmes Big Data. En effet, les architectures classiques de traitement de données basées sur l'utilisation d'une Base de Données s'avèrent indataptées aux volumes de données considérés.

\subsection{Les besoins liés au traitement de grandes masses de données}\label{ss:besoins}

Les systèmes modernes doivent traiter des données de plus en plus volumineuses. En effet, à l'heure actuelle, plusieurs gigaoctets de données sont générés dans le monde à chaque seconde. Les entreprises ont besoin de systèmes efficaces à même de traiter ces masses. Les principaux défis à résoudre sont les suivants :  

\begin{itemize}
   \item \textbf{Tolérance aux fautes :} le système doit continuer de fonctionner correctement malgré l'occurrence de pannes matérielles ou logicielles. Il doit également tolérer les fautes humaines, souvent sous-estimées lors du développement des systèmes. Les exemples se sont pourtant multipliés, dans un passé récent, d'erreurs de manipulation humaines ayant conduit à la perte de données où à l'interruption du service rendu.

   \item \textbf{Efficacité :} pour la plupart, les applications modernes nécessitent que les requêtes soient traitées très rapidement, c'est-à-dire qu'elles doivent garantir une très \emph{faible latence}. Typiquement, de nombreuses applications ont besoin que les traitements soient effectués en moins de quelques millisecondes. Par ailleurs, les systèmes modernes doivent être capables de traiter de nombreuses requêtes par seconde, c'est-à-dire de garantir un \emph{haut débit}. Typiquement, de nombreux systèmes traitent plusieurs milliers, voire dizaines de milliers, de requêtes par seconde. 
 
   \item \textbf{Passage à l'échelle :} les systèmes modernes doivent être capables de traiter un volume continuellement croissant de données et de requêtes. Par ailleurs, les temps de traitement de ces requêtes doivent être aussi peu affectés que possible par le volume des données traitées. Pour parvenir à passer à l'échelle, la seule solution possible est de construire des systèmes autorisant l'ajout dynamique de ressources matérielles sans altération des performances. 
 
   \item \textbf{Extensibilité :} le système doit permettre l'ajout ou la modification de fonctionnalités au cours du temps. Il est important de noter que ces modifications apportées au système conduisent parfois à modifier le format des données stockées (par exemple pour ajouter une information qui n'était initialement pas stockée). Il est crucial que le système permette la migration d'un format de données vers un autre à moindre effort.

   \item \textbf{Maintenabilité :} les systèmes de traitement de masses de données sont complexes. Il est primordial que leur maintenance soit aussi simple que possible. En effet, face au volume croissant des données traitées, des opérations de maintance doivent régulièrement être planifiées afin de garantir un fonctionnement continu du service. \`A titre d'exemple, une opération classique de maintenance consiste à ajuster les ressources matérielles mises à disposition afin de garantir que le système passe à l'échelle. Pour rendre celui-ci aussi maintenable que possible, il est nécessaire de clairement identifier les caractéristiques de chaque composant, afin de sélectionner les plus simples possibles à même de répondre aux exigences. Nous verrons dans ce chapitre que chaque composant d'un système de gestion de données a des exigences spécifiques. 
\end{itemize}


\subsection{Limites des architectures incrémentales}\label{ss:limites}

Les systèmes traditionnels de gestion de données ont une architecture dite \emph{incrémentale}. Comme nous le représentons sur la figure~\ref{f:architecture_incrementale}, ce qui caractérise ces architectures est l'utilisation d'une base de données pour stocker l'état du système\footnote{Il est important de noter que nous utilisons le mot bases de données au sens large. Il peut s'agir aussi bien de bases relationnelles que NoSQL.} et répondre aux requêtes de lecture sur les données. Cette base est mise à jour au fur et à mesure que des données sont reçues ou traitées. Par exemple, supposons que l'on désire développer un système qui compte le nombre de visites que chaque URL d'un site web reçoit. L'architecture incrémentale suivante peut être réalisée : on stocke dans une table de la base de données l'ensemble des URL, ainsi qu'un compteur de visites pour chacune. \`A chaque fois qu'une URL est visitée, on met à jour l'état de la base en incrémentant le compteur associé. 

%Exemple d'un système permettant d'analyser les consultations des pages d'un site Web. L'une des fonctions attendues est de pouvoir calculer le nombre de visiteurs uniques par URL sur une période de temps donnée.

%Un tel système reçoit deux types d'information : 
%\begin{itemize}
%   \item visites : contient l'identifiant d'un utilisateur, une URL et une estampille.
%   \item alias : contient deux identifiants et indique qu'il s'agit d'une seule et même personne.  
%\end{itemize}

%Dans un système incrémental, une Base de Données reçoit les mise à jour et des requêtes de lecture (consultation des données) et se met à jour de façon incrémentale à chaque requête (ajouter Schéma ?).

\begin{figure}[h]
   \centering
   \includegraphics[scale=0.5]{Figs/chapter-Gestion/incremental_architecture.pdf}
   \caption{Architecture incrémentale.}
   \label{f:architecture_incrementale}
\end{figure}

Les architectures incrémentales ont plusieurs limitations. En voici les principales : 

\begin{itemize}
   \item \textbf{Opérations de maintenance coûteuses :} la base de données doit régulièrement réaliser des opérations afin de rester opérationnelle. \`A titre d'exemple, citons les opérations de compactage qui libèrent des zones de disques qui ne sont plus utilisées suite à des mises à jour de la base. De telles opérations sont si coûteuses qu'elles peuvent engendrer une indisponibilité du système si elles ne sont pas réalisées aux moments opportuns. Il est par exemple impératif de garantir que les différents réplicas d'une base de données ne les réaliseront pas en même temps, sans quoi le système pourrait devenir inopérant.   
 
   \item \textbf{Gestion complexe de la cohérence des données :} une autre difficulté rencontrée avec les architectures incrémentales est la gestion de la cohérence des données. En effet, afin (1) de rendre le système hautement disponible et (2) de répondre à un plus grand nombre de requêtes, il est nécessaire de répliquer la base de données sur plusieurs machines physiques. Dès lors que plusieurs réplicas sont utilisés, se posent des problèmes de cohérence des données. En effet, si pour diverses raisons (panne matérielle, congestion), le réseau ne permet temporairement plus aux réplicas de communiquer, le système devra soit cesser de se mettre à jour et de répondre aux requêtes, soit accepter que l'état des réplicas diverge. Suite à une divergence temporaire, il est complexe de remettre la base de données dans un état cohérent. Par ailleurs, les réponses fournies aux requêtes pendant les périodes de divergence sont parfois erronnées, ce qui peut, dans certains cas, avoir des conséquences problématiques sur la cohérence globale du système.

   \item \textbf{Exposition aux fautes humaines :} une troisième difficulté rencontrée avec les architectures incrémentales est leur exposition aux fautes humaines. En effet, il suffit parfois d'une manipulation malencontreuse pour altérer l'état de la base. Les données risquent ainsi d'être corrompues et il n'est pas toujours aisé de détecter la cause de la corruption, ni de remettre la base dans un état cohérent. 
\end{itemize}

\section{L'architecture Lambda pour le\\traitement de grandes masses \\de données}\label{s:lambda}

Nous avons décrit dans la section précédente les besoins des systèmes qui traitent de grandes masses de données, ainsi que les limites des architectures traditionnelles, dites incrémentales. Dans cette section, nous présentons une architecture logicielle qui a été proposée pour répondre à ces besoins. Appelée Architecture Lambda, elle a initialement été proposée par Nathan Marz, créateur du logiciel \it{open source} Storm, largement utilisé dans l'industrie, notamment par le système Twitter. 

L'architecture Lambda est composée de trois couches (figure~\ref{f:lambda-architecture}) ; chacune satisfait un sous-ensemble des propriétés listées à la section~\ref{ss:besoins} et offre des fonctionnalités spécifiques. C'est la conjonction des trois couches qui permet de répondre aux besoins des systèmes qui traitent de grandes masses de données. 

%L'objectif serait de parvenir à réaliser $requete = fonction (donnees)$. Malheureusement, ce n'est pas envisageable. Supposons que l'on ait 1TB de données. Chaque requête prendrait un temps extrêmement long à être traiter.

\begin{figure}[h]
   \centering
   \includegraphics[scale=.55]{Figs/chapter-Gestion/lambda.pdf}
   \caption{L'architecture Lambda.}
   \label{f:lambda-architecture}
\end{figure}

Nous présentons brièvement les trois couches ci-après, avant d'en donner une description plus détaillée dans les sections suivantes. 

%\begin{itemize}
\paragraph{La couche \og batch~\fg{}} Il s'agit de la couche de base de l'architecture Lambda. Elle offre deux services. Tout d'abord, elle est en charge de \emph{stocker les données reçues} par le système (appelées Jeu de données principal sur la figure~\ref{f:lambda-architecture}). Ces dernières sont écrites au fur et à mesure de leur arrivée. Elles ne sont ensuite jamais mises à jour. Les uniques opérations qui sont faites sur ces données sont des requêtes en lecture. Ces données sont donc non mutables et leur volume augmente au cours du temps. D'autre part, la couche batch est en charge d'\emph{appliquer des traitement arbitraires} sur les données stockées, à l'aide de systèmes dits \guil{de traitements par lots}, c'est-à-dire des systèmes qui sont adaptés pour effectuer des calculs sur des grandes masses de données en les décomposant par lots. L'exemple le plus célèbre est le logiciel \it{open source} Apache Hadoop, largement utilisé dans l'industrie. L'intérêt principal de ces systèmes réside dans le fait qu'ils sont massivement parallèles : ils sont ainsi capables d'exploiter de façon efficace un ensemble de machines résidant dans un centre de calcul. Les traitements réalisés par la couche batch produisent des données traitées appelées vues batch. Par exemple, supposons que l'on souhaite réaliser un système analysant la distribution temporelle du nombre de clics sur les différentes URL d'un site web. La couche batch stockerait dans le jeu de données principal les informations brutes correspondant à chaque clic (quel utilisateur ? sur quelle page ? à quel moment ?) et une vue batch pourrait être régulièrement produite pour calculer le nombre de clics sur chaque URL par intervalle de temps d'une heure. Notons que cet exemple illustre le fait que la couche batch n'a pas toujours besoin de réaliser les traitements sur tout le jeu de données principal. En effet, il est possible dans ce cas d'appliquer des algorithmes de traitements par lots incrémentaux, c'est-à-dire qui ne traitent que les dernières données reçues depuis leur précédente exécution. %s masses de données lues A noter que ces calculs ne sont pas toujours fait à partir de 0. Il est possible, dans certains cas, de faire des algorithmes incrémentaux. Par ailleurs, ces traitements peuvent la plupart du temps être faits de façon massivement parallèle. Par exemple, dans le cas des URL, il suffit de traiter de façon parallèle des ensembles d'URLs différents.

\paragraph{La couche \og service~\fg{}} La couche service a deux objectifs~; le premier est de \emph{stocker les vues batch}, le second est de \emph{répondre à des requêtes aléatoires en lecture sur les vues batch}. Cette couche est implantée à l'aide d'une base qui doit être efficace pour réaliser les deux opérations suivantes : mise à jour de la base de données par lots (c'est-à-dire écriture d'une vue en un bloc) et lectures aléatoires des données stockées dans les vues. 
      
\paragraph{La couche \og vitesse~\fg{}} La couche vitesse a trois objectifs. Le premier est de \emph{traiter les données qui ne l'ont pas encore été par la couche batch} (et qui ne sont donc pas représentées dans les vues batch stockées par la couche service). La couche service est nécessaire car les calculs effectués par la couche batch peuvent durer plusieurs minutes, voire heures. Ainsi, certaines données (celles reçues après le démarrage des traitements par lots exécutés par la couche batch) ne sont pas prises en compte dans les vues batch. La couche vitesse produit elle aussi des données traitées, appelées \it{vues temps réel}. Afin de minimiser la latence de traitement des données, les traitements réalisés par la couche vitesse ne le sont pas par lots, mais au fil de l'eau, au fur et à mesure que des données sont reçues. On parle ainsi de traitements de flux de données. Ils sont, par nature, incrémentaux. Le deuxième objectif de la couche vitesse est de \emph{stocker les vues temps réel} produites par les traitements de flux de données. Enfin, le troisième objectif est de \emph{répondre à des requêtes aléatoires en lecture sur les vues temps réel}. La base de données en charge du stockage des vues temps réel doit donc être efficace à la fois pour les écritures aléatoires et les lectures aléatoires. 

         %$realtime\_view = function (realtime\_view, nouvelles donnees)$

%\end{itemize}

Enfin, comme illustré sur la figure~\ref{f:lambda-architecture}, les requêtes exploitent les couches service et vitesse afin d'obtenir à la fois les données traitées par lots (stockées dans les vues batch) et celles traitée au fil de l'eau (stockées dans les vues temps réel). Ainsi, on peut résumer l'architecture Lambda à l'aide des trois équations suivantes :

\begin{itemize}
   \item $vues\_batch = fonction(jeu\_de\_donn\acute{e}es\_principal)$
   \item $vues\_temps\_r\acute{e}el = fonction(vues\_temps\_r\acute{e}el,~nouvelles\_donn\acute{e}es)$
   \item $requ\hat{e}te = fonction(vues\_batch,~vues\_temps\_r\acute{e}el)$
\end{itemize}


\section{La couche batch~: traitement de \\ données par lots}

Comme expliqué dans la section précédente, la couche batch est la principale de l'architecture Lambda. Elle est en charge du stockage du jeu de données principal, ainsi que des traitements par lots sur ces données. Dans cette section, nous abordons trois sujets : les caractéristiques du jeu de données principal, le système de stockage de ces données et les techniques de traitement par lots.

%Il y a deux questions clés qui se posent concernant les données stockées par la couche "batch" : d'une part, le modèle de données qui est utilisé ; d'autre part, la façon dont les données sont stockées.

\subsection{Caractéristiques du jeu de données\\principal}

L'objectif d'un système de gestion de données à large-échelle est de répondre à différentes questions relatives aux données stockées. Une particularité de ces systèmes est que l'ensemble des questions auxquelles il conviendra tôt ou tard de répondre n'est souvent pas connu à l'avance. Afin de ne pas restreindre les possibilités, il est primordial que le jeu de données principal stocké dans la couche batch soit le plus \emph{brut} possible. En effet, effectuer un traitement des données préalablement à leur stockage peut engendrer une perte d'information et s'avérer préjudiciable. Prenons l'exemple d'un système de gestion de données boursières. Quand bien même les statistiques seraient réalisées à l'échelle de l'heure, il est important de conserver des données à la minute (si disponibles) afin de pouvoir, si nécessaire, changer la granularité à laquelle les événements seront analysés.

Il est par ailleurs important que le jeu de données principal ne soit \emph{pas mutable}. Prenons le cas de données qui pourraient être stockées pour un réseau social. Supposons qu'à tout membre de ce réseau social soit associé un lieu de résidence. En cas de déménagement d'un membre, il n'est pas souhaitable de mettre à jour (et donc d'écraser) l'information correspondant à sa résidence. Il est en effet préférable de conserver comme information le fait qu'il a eu une première adresse, jusqu'à une date donnée, à partir de laquelle il a eu un nouveau lieu d'habitation. En effet, cette information pourra être exploitée plus tard, par exemple, pour dresser des statistiques sur les migrations des membres du réseau social, ou encore réaliser des analyses complexes comme \og quelles sont les attentes des personnes ayant déménagé d'une zone de campagne vers une zone urbaine~\fg{}. Cet exemple illustre également le fait qu'il est souvent pertinent d'associer à une donnée une \emph{estampille} indiquant à quelle date précise elle a été produite. 

Il est bien entendu possible de mettre en place un mécanisme de \emph{ramasse-miettes} visant à supprimer des données considérées inutiles. Par exemple, si l'on reprend l'exemple du réseau social, il est tout à fait possible que son responsable décide, à un moment donné, que les informations sur les changements de résidence ne seront jamais exploitées. Dans ce cas, il est pertinent, afin de limiter la quantité de données à stocker de ne conserver que la dernière résidence de chaque membre. 

\subsection{Stockage du jeu de données principal}

Le jeu de données principal n'est pas mutable et a une taille croissante au cours du temps. Par ailleurs, il est complété au fur et à mesure que des données sont reçues et il est lu de façon périodique, par lots, par le système de traitement de données. Ainsi, les attentes concernant le stockage du jeu de données principal sont les suivantes : 

\begin{itemize}
\item \textbf{\'Ecritures incrémentales :} les données stockées dans la couche batch ne sont pas mutables. En conséquence, chaque information va être écrite une seule fois. Le système de stockage ne doit donc pas fournir un support efficace pour la mise à jour des données. Il doit en revanche supporter de façon efficace l'ajout incrémental de données.

\item \textbf{Lectures par lots :} pour permettre l'exécution des algorithmes de traitements, le système de stockage doit permettre de lire efficacement des lots de données. En revanche, il n'a pas besoin de fournir un support efficace pour la lecture de petits ensembles de données aléatoirement choisis. 

\item \textbf{Passage à l'échelle :} le système doit être en mesure de stocker l'intégralité du jeu de données principal. Ce dernier croissant au cours du temps, il est impératif de pouvoir augmenter la capacité de stockage par ajout de machines, sans affecter les performances de lecture et d'écriture. 
 
\item \textbf{Tolérance aux fautes :} le système de stockage doit supporter des pannes matérielles, que ce soit, par exemple, d'une machine ou encore d'un disque. 
\end{itemize} 
 
Pour stocker le jeu de données principal, l'option la plus pertinente est souvent d'utiliser un système de fichiers distribué. Afin d'illustrer le fonctionnement d'un tel système, prenons l'exemple de l'un des plus populaires d'entre eux, à savoir HDFS (Hadoop Distributed File System). HDFS s'exécute sur une grappe de machines. L'une de ces dernières est appelée \it{namenode}~; les autres sont les \it{datanodes}. Quand un fichier est stocké dans HDFS, il est décomposé en blocs de taille fixe (typiquement entre 64~Mo et 256~Mo). Chaque bloc est répliqué sur plusieurs datanodes\footnote{HDFS est souvent déployé avec un facteur de réplication de 3, ce qui signifie que chaque bloc est stocké sur 3 datanodes différents.}. Le rôle du namenode est de stocker l'information sur les associations fichier-blocs et sur la localisation des blocs. Ainsi, quand un programme a besoin d'accéder à un fichier stocké dans HDFS, il contacte le namenode\index{namenode} pour connaître la localisation des différents blocs. Il accède ensuite directement aux datanodes qui stockent les blocs requis.

Le stockage des données de la couche batch au sein du système HDFS peut être réalisé ainsi : à chaque fois que de nouvelles données sont produites, de nouveaux fichiers sont créés pour les contenir. Afin d'améliorer l'efficacité des traitements qui seront appliqués sur ces fichiers, il peut être intéressant de procéder à ce que l'on appelle un \emph{partitionnement vertical} de ces derniers. Ce mode de partitionnement consiste à créer des répertoires dans le but de dinstinguer facilement des groupes de données. \`A titre d'exemple, il est possible de créer un nouveau répertoire par jour. Ainsi, il sera facile d'appliquer des fonctions de traitement sur des sous-ensembles de données, telles que \og les données des deux dernières semaines~\fg{}.

\subsection{Traitement de données par lots}

La couche batch doit permettre le traitement du jeu de données principal. Pour des raisons d'efficacité, il est préconisé d'utiliser un système de traitement par lots. Dans cette section, nous débutons par quelques remarques préliminaires sur ces systèmes. Nous décrivons ensuite le paradigme Map-Reduce, largement utilisé dans l'industrie. 
 
\subsubsection{Remarques préliminaires sur le traitement de données par lots}

L'objectif des traitements réalisés par la couche batch est de produire des vues qui seront ensuite exploitées par la couche service pour répondre aux requêtes des clients. Considérons l'exemple d'un système permettant de calculer le nombre de visiteurs uniques par URL sur une période de temps donnée. 

La première question qui se pose est : quelles vues fournir à la couche service ? Dans l'objectif de rendre cette dernière aussi rapide que possible, il serait logique de multiplier les vues. Ainsi, dans l'exemple considéré, nous pourrions imaginer qu'une vue soit créée pour chaque période de temps imaginable. Cette approche est toutefois impossible à mettre en \oe{}uvre. En effet, supposons que nous ayons des données sur un an, il faudrait plusieurs centaines de millions de vues pour être en mesure de disposer de toutes les périodes de temps imaginables. En revanche, une solution envisageable est de calculer des vues pour chaque période d'une heure (ce qui représente 8760 vues par an). Il appartient ensuite à la vue service de calculer la somme des heures correspondant à une période de temps donnée.

La seconde question qui se pose est de déterminer s'il faut utiliser un algorithme de traitement \emph{incrémental} ou \emph{intégral}. Le premier type conserve les vues calculées dans le passé et les met à jour avec les nouvelles données (traitées par lots). Le second type d'algorithme traite, à chaque itération, l'intégralité des données et remplace les anciennes vues par de nouvelles vues. Les deux types ont des avantages et des inconvénients~: 

\begin{itemize}
   \item Les ressources requises pour l'exécution sont souvent moindres dans le cas d'un algorithme incrémental. En revanche, les vues sont souvent de plus grande taille. Supposons, par exemple, que l'algorithme crée des vues contenant des valeurs moyennes. Il sera nécessaire, dans le cas d'algorithmes incrémentaux, de stocker, outre la moyenne, le nombre de données prises en compte dans la moyenne, sans quoi cette dernière ne pourra pas être mise à jour de façon incrémentale. 
   
   \item Les algorithmes incrémentaux sont plus sensibles aux fautes humaines que les intégraux. En effet, supposons que l'algorithme employé pour effectuer un traitement ait un bogue. Dans le cas d'un algorithme intégral, il suffit de supprimer le bogue et de relancer le calcul sur l'intégralité des données. Dans le cas d'un algorithme incrémental, il est souvent très difficile (voir impossible) de corriger les erreurs introduites dans la vue par les itérations réalisées alors que le bogue était présent. La seule possibilité est alors d'utiliser un algorithme intégral pour reconstruire les vues.

   \item Les algorithmes incrémentaux sont souvent plus complexes que les intégraux. Ceci vient du fait qu'ils doivent traiter à la fois des données brutes et d'autres traitées et stockées au sein des vues. Par opposition, les algorithmes intégraux ne traitent que des données brutes.  
\end{itemize}

Enfin, notons qu'il est crucial que les algorithmes de traitement de données par lots \emph{passent à l'échelle}. En effet, nous avons vu que la taille des données stockées par la couche batch est strictement croissante. Il convient donc de définir des algorithmes de traitement dont le temps d'exécution peut rester quasiment constant, malgré l'augmentation de la taille des données, en procédant à des ajouts de ressources matérielles pour leur exécution. 

\subsubsection{Le paradigme Map-Reduce}

Map-Reduce est un paradigme de programmation de calculs distribués proposé par Google en 2004. Il permet d'écrire des algorithmes de traitement par lots tolérant les fautes et passant à l'échelle. Il est donc tout à fait adapté pour l'écriture des algorithmes de traitement de la couche batch, ce qui explique sa très large adoption par l'industrie au cours de la dernière décennie.

Le paradigme Map-Reduce propose d'écrire les algorithmes sous forme de fonctions $map$ et $reduce$ manipulant des paires clé/valeur. Il est très expressif et permet donc d'implanter de très nombreuses fonctions. L'exemple très souvent donné pour illustrer Map-Reduce est celui d'un algorithme comptant le nombre d'occurences de chaque mot d'un texte. La fonction $map$ émet des paires clé/valeur pour chaque mot rencontré dans le texte. La clé correspond au mot, tandis que la valeur est positionnée à 1. Le code de la fonction $map$ prend ainsi la forme suivante (le texte est représenté par le paramètre $sentence$) :\\ 

\begin{verbatim}
   function word_count_map(sentence) {
      for (word in sentence.split(" ")) {
         emit(word, 1)
      }
   }
\end{verbatim}

La fonction $reduce$ prend en entrée l'ensemble des valeurs associées à une même clé et effectue un traitement dessus. Dans le cas considéré (compter le nombre d'occurences de chaque mot d'un texte), la fonction \it{reduce} doit simplement additionner les valeurs reçues. Son code prend ainsi la forme suivante :\\   

\newpage

\begin{verbatim}
   function word_count_reduce(word, values) {
      sum = 0
      for (val in values) {
         sum += val
      }
      emit(word, sum)
   }
\end{verbatim}

Les canevas logiciels implantant le paradigme Map-Reduce prennent en charge toute la complexité d'exécution du code écrit précédemment. \`A titre d'exemple, ils découpent automatiquement les données en entrée en blocs de données de même taille, puis planifient l'exécution des tâches sur les machines disponibles au sein de la grappe. Par ailleurs, les canevas offrent des garanties en termes de tolérance aux fautes en implantant, par exemple, des mécanismes capables de reaffecter des calculs ayant échoué. Enfin, notons que l'exemple donné pour illustrer le paradigme est volontairement très simple ; il est possible de réaliser des algorithmes très complexes à l'aide de Map-Reduce. Nous invitons les lecteurs intéressés à lire l'un des nombreux livres dédiés à ce sujet. 

%\subsection{Requêtes sur les données traitées}

%\begin{figure}[h]
%   \centering
%   \includegraphics[scale=.45]{Figs/chapter-Gestion/map_reduce.pdf}
%   \caption{Exemple de calcul Map-Reduce.}
%   \label{f:map-reduce}
%\end{figure}



\section{La couche service : stockage \\ et requêtes sur les vues batch}

Comme nous l'avons expliqué à la section~\ref{s:lambda}, la couche service a deux objectifs. D'une part, elle est en charge de stocker les vues batch produites par les traitements par lots. D'autre part, elle doit répondre à des requêtes aléatoires en lecture sur les vues batch. La couche service est donc naturellement mise en \oe{}uvre par une base de données. Nous débutons cette section par une remarque préliminaire sur le stockage des vues batch. Nous décrivons ensuite les caractéristiques que doit présenter la base de données de la couche service.
 
\subsection{Remarque préliminaire sur le stockage \\des vues batch}

%\textbf{TODO : expliquer que l'on a parlé de stockage aussi bien dans la couche "batch" que "service", mais que l'on a, jusqu'à présent, occulté une question qui est celle de la normalisation ou non des données.} 

Dès lors que l'on considère le stockage de données, se pose la question de \emph{normaliser} ou non les données. Pour expliquer ce concept et déterminer le choix à réaliser dans le cadre du stockage de données dans l'architecture Lambda, prenons un exemple concret : supposons que l'on souhaite stocker des données sur la localisation d'individus. Deux stratégies sont possibles. Une première stratégie, traditionnellement utilisée dans les bases de données relationnelles, consiste à \emph{normaliser} les données (figure~\ref{f:normalisation}). Dans cet exemple, cela consisterait à créer une table pour stocker les différentes localisations associées chacune à un identifiant. Une autre table stockerait l'ensemble des individus, en précisant, pour chacun d'entre eux, sa localisation à l'aide de l'identifiant associé. Cette stratégie de normalisation a pour avantage de ne pas dupliquer le stockage de données potentiellement volumineuses (par exemple, dans le cas considéré, stocker une localisation peut consommer beaucoup d'espace car nécessitant de préciser pays, ville, rue). Cette stratégie limite également les erreurs possibles car, dès lors qu'une information est stockée de façon redondante (i.e. les localisations dans notre exemple si l'on n'utilise pas de stratégie de normalisation), il est nécessaire de s'assurer que les différentes versions sont cohérentes. 

\begin{figure}[h]
   \centering
   \includegraphics[scale=.45]{Figs/chapter-Gestion/normalisation.pdf}
   \caption{Exemple de stockage normalisé.}
   \label{f:normalisation}
\end{figure}


La stratégie de normalisation présente toutefois un inconvénient : il est nécessaire de réaliser des opérations de jointure entre tables pour obtenir la réponse à certaines requêtes. Typiquement, dans le cas présenté précédemment, il est nécessaire de réaliser une jointure entre les deux tables créées pour déterminer la localisation d'un utilisateur. Cette opération est coûteuse et, si l'application nécessite de requêter fréquemment la localisation d'un utilisateur, il peut s'avérer pertinent de supprimer la nécessité d'effectuer une jointure. Pour ce faire, la seule solution est de stocker, pour chaque utilisateur, les données complètes de localisation (et non un identifiant). Cela consiste à \emph{dénormaliser} les données. Cette stratégie est illustrée sur la figure~\ref{f:denormalisation}.

\begin{figure}[h]
   \centering
   \includegraphics[scale=.45]{Figs/chapter-Gestion/denormalisation.pdf}
   \caption{Exemple de stockage dénormalisé.}
   \label{f:denormalisation}
\end{figure}

Dans les architectures Lambda, il est classique d'employer une stratégie de normalisation au sein de la couche batch et une stratégie de dénormalisation au sein de la couche service. En effet, les données stockées dans la couche batch n'ont pas besoin d'être requêtées de façon aléatoire (les lectures sont réalisées par lots). Les avantages apportés par la normalisation prennent donc le dessus : volume de stockage moins important, pas de problème de cohérence entre les différentes répliques d'une donnée à traiter. En revanche, au sein de la couche service, les données peuvent être requêtées de façon arbitrairement complexe et il est important de garantir que les meilleures performances seront atteintes. Afin de limiter des opérations coûteuses, comme les jointures, il est préférable d'utiliser un stockage dénormalisé des données. 
 
\subsection{Stockage des vues batch}

%Le jeu de données principal est immutable et a une taille croissante au cours du temps. Par ailleurs, il est mis à jour au fur et à mesure que des données sont reçues et il est lu de façon périodique, par lots, par le système de traitement de données par lots. Ainsi, 

Les attentes concernant le stockage des vues batch sont les suivantes : 

\begin{itemize}
\item \textbf{\'Ecritures par lots :} les vues batch stockées dans la couche service sont écrites par lots par les algorithmes de traitement s'exécutant au sein de la couche batch. Il est souvent nécessaire de remplacer l'intégralité d'une vue batch par sa nouvelle version. Il est en revanche inutile de gérer efficacement les écritures aléatoires de données. 

\item \textbf{Lectures aléatoires :} le système de stockage doit garantir l'efficacité des lectures aléatoires de données, dans le but de répondre aussi rapidement que possible aux requêtes des clients. Il est ainsi impératif que les données soit indexées de façon à y accéder très rapidement.  

\item \textbf{Passage à l'échelle :} le système de stockage doit être en mesure de stocker des vues batch de tailles arbitraires, souvent croissantes avec le temps, et de répondre à une charge croissante de requêtes clientes. Il est donc impératif de pouvoir augmenter sa capacité en procédant à des ajouts de machines physiques, tout en minimisant les pertes de performance. 
 
\item \textbf{Tolérance aux fautes :} le système de stockage doit supporter des pannes matérielles, que ce soit, par exemple, d'une machine ou encore d'un disque. 
\end{itemize} 
 
Il est très important de noter que le système de stockage des vues batch au sein de la couche service ne requiert pas de supporter les écritures aléatoires. En effet, ces écritures sont la principale source de complexité pour concevoir une base de données efficace. \`A titre d'exemple, les algorithmes de compactage mentionnés à la section~\ref{ss:limites} sont beaucoup plus simples dès lors qu'il n'y a pas nécessité de supporter des écritures aléatoires. \'Egalement à titre d'exemple, ne pas autoriser les écritures aléatoires simplifie grandement la gestion de la synchronisation entre les requêtes concurrentes sur la base de données (qu'elles soient en lecture ou en écriture).

Un exemple populaire de gestionnaire de bases de données employé dans la couche service est ElephantDB, qui a été spécialement conçu pour stocker des paires clé/valeur produites par Hadoop, un canevas populaire implantant le paradigme Map-Reduce. 


\section{La couche vitesse : traitement de flux de données, stockage et requêtes sur les vues temps réel}
%\section{Traitement distribué de type flux de données}
%\subsection{Vues temps-réelles des données}
%\subsection{Traitement temps-réel de flux de données}

Comme nous l'avons expliqué précédemment, la couche vitesse a trois objectifs. Le premier est de traiter les données qui ne l'ont pas encore été par la couche batch. En effet, cette dernière traite les données dans leur intégralité, ce qui engendre des temps de calcul importants. Pour prendre en compte les données récentes, la couche vitesse applique au fil de l'eau des traitements \emph{incrémentaux}, appelés traitements de flux de données, qui produisent des vues temps réel. Le second objectif de la couche vitesse est de stocker ces dernières. Enfin, son troisième objectif est de répondre à des requêtes aléatoires en lecture sur les vues temps réel. 

Dans cette section, nous débutons par la description des systèmes traitant les flux de données. Nous décrivons ensuite les caractéristiques attendues des systèmes de stockage des vues temps réel.   

%La couche "service" est donc naturellement mise en \oe{}uvre par une base de données. Nous débutons cette section par une remarque préliminaire sur le stockage des vues "batch". Nous décrivons ensuite les caractéristiques que doit avoir la base de données de la couche "service".

%
%Le pourquoi du calcul incrémental : 
%\begin{itemize}
%\item Les vues temps réelles ne représentent que les dernières données et doivent être mises à jour très peu de temps après que chaque nouvelle donnée est reçue. Typiquement, les latences souhaitées sont de quelques millisecondes à quelques secondes.
%\item Tout d'abord, notons le fait qu'il n'est pas possible de faire des calculs intégraux sur l'ensemble des données récentes. La raison est la suivante. Supposons qu'un système produise 32GB de données par jour et que la vue batch soit mise à jour toutes les 6 heures. Cela siginifie que la vue "vitesse" devra traiter 6h de données, ce qui représente 8GB. Si les données ont une taille moyenne de 100 octets, cela signifie que 86 millions de données sont reçues en 6h. Faire un calcul 86 millions de fois sur un ensemble de données pouvant représenter plusieurs Go n'est tout simplement pas réalisable avec une latence inférieure à la minute.  
%\end{itemize}

\subsection{Traitement de flux de données}

L'architecture typique d'un système traitant les flux de données est représentée sur la figure~\ref{f:queues_workers}. Ces systèmes reposent sur deux types d'entités : 

\begin{itemize}
\item \textbf{Les queues :} ce sont des files de messages qui permettent de stocker momentanément les données.
\item \textbf{Les workers :} ce sont des processus de traitement de données. Chacun consomme les messages stockés dans les queues, les traite et les place dans la queue suivante. Les derniers processus à intervenir ont la charge d'écrire les données traitées dans une base, peuplant ainsi des vues dites temps réel. 
\end{itemize}

\begin{figure}[h]
   \centering
   \includegraphics[width=\textwidth]{Figs/chapter-Gestion/stream_processing.pdf}
   \caption{Architecture d'un système de traitement de flux de données.}
   \label{f:queues_workers}
\end{figure}

Afin d'illustrer cette architecture, prenons l'exemple d'un algorithme qui compte le nombre de visites que chaque URL d'un site web reçoit. Un tel algorithme peut être implanté à l'aide d'un \it{pipeline} de quatre étages : les premier et troisième étages sont constitués de queues, alors que les deuxième et quatrième sont constitués de workers. Les queues du premier étage stockent les données sur les visites. Ces queues sont mises à jour à chaque fois qu'une URL est visitée. Les workers du deuxième étage consomment les données du premier étage et filtrent celles qui sont incorrectes (i.e. les données sur les visites ayant une URL incorrecte). Les données correctes sont écrites dans les queues du troisième. Enfin, les workers du quatrième étage traitent les données stockées dans le troisième étage. Le traitement consiste à mettre à jour une base de données contenant le décompte des visites pour chaque URL. 


\subsection{Stockage des vues temps réel}

Les données traitées par le système précédent sont stockées par la couche vitesse dans une base. Cette dernière doit avoir les caractéristiques suivantes : 

\begin{itemize}
\item \textbf{Lectures aléatoires :}  le système de stockage doit permettre d'effectuer efficacement des lectures aléatoires des données, afin de répondre le plus rapidement possible aux requêtes des clients. Cela signifie, en outre, que les données stockées doivent être indexées.
 
\item \textbf{\'Ecritures aléatoires :} comme nous l'avons décrit précédemment, les données sont écrites au fur et à mesure dans la base par les algorithmes de traitement de flux. Le système de stockage doit donc garantir l'efficacité des écritures aléatoires. 

\item \textbf{Passage à l'échelle :} le système de stockage doit être capable de supporter une croissance des taux de requêtes en lecture et écriture. Dans certains cas, il est également nécessaire de supporter une croissance du volume des données stockées. Cela impose de distribuer le système de stockage sur plusieurs machines. 
 
\item \textbf{Tolérance aux fautes :} le système de stockage doit supporter des pannes matérielles, que ce soit, par exemple, d'une machine ou encore d'un disque. 
\end{itemize} 

Ces caractéristiques sont offertes par les bases de données dites \it{NoSQL} (pas seulement SQL). Ces bases offrent un paradigme différent de celui offert par les bases relationnelles classiques. Leur objectif est de permettre la manipulation de volumes de données importants, tout en autorisant un passage à l'échelle par ajout de machines. Il existe différents types de bases de données NoSQL. Ces derniers diffèrent par les modèles de données qu'ils stockent, ainsi que par les requêtes (lectures et écritures) qu'ils permettent de réaliser. Citons, à titre d'exemple, les bases de données orientées documents, clé/valeur, graphes, ou encore séries temporelles.    


\section{Conclusion}

Dans ce chapitre, nous avons étudié les systèmes de gestion de données large-échelle. Nous avons tout d'abord présenté les besoins de ces systèmes et expliqué les raisons pour lesquelles les architectures traditionnelles ne sont pas adaptées pour les réaliser. Nous avons ensuite décrit l'architecture Lambda, composée de trois couches : batch, service et vitesse. Nous avons décrit le rôle de chacune des couches et expliqué les caractéristiques qu'elles doivent présenter, en termes de performance et de tolérance aux fautes.  
 
