La « vélocité » est un des multiples V qui caractérisent les données massives. Dans de nombreux cas, de nouvelles données arrivent sous forme de flux et doivent être traitées dans des délais cohérents avec le(s) flux. Cela impose de nouvelles exigences sur la latence des opérations de construction de modèle ou de prise de décision grâce à un modèle. Par latence on entend ici l’intervalle de temps entre le moment où le traitement démarre et le moment où il se termine.
Par flux on entend une succession de données de même type, arrivant à intervalles constants ou variables. Pour être traitées, ces données peuvent être découpées en « tranches » (slices), soit à partir des intervalles d’arrivée des données, soit à partir de contraintes liées au traitement (intervalles de durée fixée, tranches de volume fixé, etc.).
Les traitements réalisés avec des données en flux sont synchrones (« continus ») ou asynchrones. Les traitements synchrones doivent être réalisés en temps réel ou, du moins, leur latence doit être compatible avec le rythme du flux. Un premier type de traitement synchrone consiste à appliquer régulièrement sur les données du flux un modèle existant, pré-estimé, afin de caractériser l’évolution du flux. Dans un second type de traitement synchrone, on estime (ou on met à jour) un modèle régulièrement à partir des données du flux, en général les dernières tranches ; ce modèle est ensuite appliqué aux données du même flux ou d’un autre flux. Lorsqu’elle est faite à partir de données en flux, la modélisation doit être incrémentale : il faut pouvoir mettre à jour le modèle existant sans avoir à le ré-estimer complètement avec toutes les données accumulées.
Les traitements asynchrones répondent à des demandes uniques ou peu fréquentes et non synchronisées avec le rythme du flux ; ils sont appliqués soit à la totalité des données du flux (si celles-ci ont été stockées), soit à des résumés de ces données. Ces traitements peuvent être coûteux mais sont en général moins contraignants en termes de latence.
Il faut noter que la fouille de flux de données ne mène pas nécessairement à une problématique spécifique aux données massives. On considère que les traitements relèvent de la fouille de données massives de flux si, en raison à la fois du débit très élevé du flux de données, du coût des traitements à appliquer à ces données et des contraintes sur la latence, il n’est pas envisageable de traiter les données régulièrement de façon exhaustive sur une plate-forme centralisée.
Réseaux de capteurs industriels. Les capteurs mesurent vibrations, températures ou noxes pour optimiser le fonctionnement des machines ou prévenir les pannes. La volumétrie peut être très élevée : les capteurs d’un seul moteur d’avion de ligne génèrent environ 1 To de données par heure de vol. Des réseaux de capteurs sont également déployés pour mesurer le trafic routier ou la densité de polluants atmosphériques dans les villes.
Logs de moteurs de recherche. Ces données (requêtes, adresses IP, cookies, etc.) permettent de cibler des publicités, de prévoir l’évolution d’épidémies ou d’estimer l’impact d’événements. Pour certaines applications, la latence doit être très faible. Google reçoit de l’ordre de 50 000 requêtes par seconde.
Sites de partage multimédia. Les méta-données accompagnant des images et vidéos (géolocalisation, date, etc.) permettent l’estimation de popularité, la détection d’événements ou le filtrage de contenus. 300 heures de vidéo sont mises en ligne chaque minute sur YouTube.
Vidéosurveillance. Les flux vidéo et audio doivent permettre la détection d’événements localisés ainsi que des corrélations entre caméras. Londres compte plus de 500 000 flux continus de vidéosurveillance.
Imagerie satellite et aérienne. Ces flux, en fort développement, proviennent de satellites en orbite héliosynchrone ou de drones professionnels. Les applications incluent le suivi de catastrophes et la détection d’événements. Les deux satellites Pléiades (résolution 50 cm) envoient 2 To de données par jour.
Architecture d’un système de traitement de flux de données¶
Le traitement d’un flux de données inclut la création et le maintien de résumés du flux, l’application de modèles et la construction et mise à jour de modèles. La figure suivante montre l’architecture générale d’un tel système.
Un système de traitement de flux doit pouvoir répondre à deux types de requêtes :
Requêtes continues, qui impliquent des traitements synchrones avec le flux. Par exemple, obtenir le nombre de requêtes par heure concernant une thématique spécifique dans un flux de requêtes d’un moteur de recherche. Les réponses constituent un flux de données de sortie.
Requêtes asynchrones, qui impliquent des traitements asynchrones par rapport au flux. Par exemple, obtenir la concentration moyenne de dioxyde d’azote sur les trois derniers jours à Paris à partir du flux d’un réseau de capteurs de pollution.
Les données du flux sont parfois conservées sur un stockage de masse à des fins d’archivage, mais les données archivées ne sont pas utilisées dans des traitements synchrones car l’accès au stockage de masse est trop lent pour être compatible avec les contraintes de latence.
Si un flux a un débit élevé, employer la totalité de ses données pour répondre à des requêtes n’est pas possible en général. L’objectif de l’échantillonnage est d’extraire du flux un échantillon suffisamment représentatif pour permettre de répondre aux différentes requêtes prévues, et d’assez faible volume pour être conservé durablement en stockage rapide.
En général, chaque donnée (ou observation) d’un flux comporte des valeurs pour plusieurs attributs (ou variables). Par exemple, le fichier de log d’un moteur de recherche contient (entre autres) l’adresse IP de l’utilisateur, une étiquette temporelle, les mots clés de la recherche, le type de navigateur employé, etc. Le flux est une succession de requêtes, dans l’ordre dans lequel elles ont été enregistrées dans le log du moteur de recherche.
L’échantillonnage simple considère une même probabilité de sélection \(p_s\) pour chaque donnée du flux, indépendamment des valeurs de ses attributs. Dans quelle mesure un tel échantillonnage permet de répondre aux requêts ? Considérons l’exemple d’un log de moteur de recherche, où l’on cherche à estimer le pourcentage des requêtes d’un utilisateur typique qui sont répétées durant un mois (voir aussi [LRU11]). Supposons que la probabilité de sélection est \(p_s = 0,1\) (nous souhaitons avoir un échantillon de 10% des requêtes) et que cette sélection est faite sans tenir compte des valeurs des attributs (adresse IP, étiquette temporelle, mots-clés de la recherche, etc.). Supposons que l’utilisateur typique fait en moyenne \(s\) requêtes non répétées par mois, \(d\) requêtes répétées une fois et un nombre négligeable de requêtes répétées plusieurs fois. Sur les \(s\) requêtes non répétées dans le flux, \(s/10\) seront présentes (une fois) dans l’échantillon. Sur les \(d\) requêtes répétées (une fois) dans le flux, seulement \(1/100\)\((= \frac{1}{10} \cdot \frac{1}{10}\)) seront répétées (une fois) dans l’échantillon. En effet, chaque requête répétée dans le flux a deux occurrences, chacune de ces occurrences a une probabilité de \(\frac{1}{10}\) d’être retenue dans l’échantillon (\(p_s = 0,1\)), pour que la requête soit répétée dans l’échantillon il faut que chacune de ses deux occurrences soit retenue et les tirages correspondants sont indépendants. Aussi, sur les \(d\) requêtes répétées une fois dans le flux, \(18/100\)\((= \frac{1}{10} \cdot \frac{9}{10} + \frac{9}{10} \cdot \frac{1}{10})\) seront présentes dans l’échantillon une seule fois (soit la première occurrence est présente et la seconde absente, soit la première est absente et la seconde présente). A partir de l’échantillon nous obtenons donc une estimation \(\frac{\frac{d}{100}}{\frac{s}{10}+\frac{18 d}{100}+\frac{d}{100}} = \frac{d}{10 s + 19 d}\), très éloignée de la valeur obtenue à partir de toutes les requêtes, qui est de \(\frac{d}{s + d}\). La solution consiste à sélectionner dans l’échantillon non 10% des requêtes mais plutôt 10% des utilisateurs, pour lesquels on conserve toutes les requêtes ; ici encore, le volume de données de l’échantillon reste le même (10% du flux) mais l’échantillon obtenu en sélectionnant les utilisateurs nous donne la possibilité de répondre à la requête. Par ailleurs, cet échantillon de 10% des utilisateurs permet de répondre également à d’autres questions concernant le comportement des utilisateurs.
Pour pouvoir répondre à des requêtes à partir de l’échantillon, il est donc souvent nécessaire de tenir compte des valeurs d’un ou plusieurs attributs, dont le choix dépend des requêtes envisagées, pour réaliser l’échantillonnage. La sélection peut être réalisée soit par la recherche de la valeur dans une liste, soit en utilisant une méthode de hachage. On définit une fonction de hachage sur l’ensemble des valeurs possibles de l’attribut visé, avec \(1/p_s\) valeurs de hash différentes ; on conserve une valeur de l’attribut si son hash est égal à une valeur fixée a priori. Cette solution de hachage permet également un réglage fin de \(p_s\) (sans contraindre \(1/p_s\) à être entier) et une variation dans le temps du pourcentage d’échantillonnage si le débit augmente.
Il est souvent nécessaire d’appliquer des traitements particuliers aux données d’un flux qui satisfont certaines conditions. Lorsque la condition de filtrage est l’appartenance de la valeur d’un attribut à un très grand ensemble de valeurs, le coût d’une vérification par comparaison directe (table de hachage classique, par exemple) peut être excessif en termes de temps et de capacité mémoire.
Un filtre de Bloom est une structure de données probabiliste en mémoire qui permet de réaliser ce filtrage efficacement. La figure suivante illustre son fonctionnement.
On dispose d’une zone mémoire de \(N\) bits, initialisée à 0. Une ou plusieurs fonctions de hachage (\(k\) fonctions) sont appliquées à chaque valeur de l’ensemble à filtrer, et les bits identifiés par les hash obtenus sont positionnés à 1 (p1, p2, p3, p4 dans la Fig. 104). Pour vérifier si une valeur nouvelle appartient à l’ensemble à filtrer, on calcule ses \(k\)hash : si au moins un des bits correspondants est à 0 (n1 dans la Fig. 104), la valeur n’appartient pas à l’ensemble ; il n’y a donc pas de faux négatifs, par construction. Si tous les bits sont à 1, la valeur est considérée appartenir à l’ensemble ; des faux positifs sont possibles en cas de collision avec les hash d’autres valeurs appartenant à l’ensemble (comme dans la Fig. 104 entre fp1 et p1).
On peut montrer que, pour \(M\) valeurs à filtrer et \(k\) fonctions de hachage avec chacune \(N\) valeurs de hash différentes, le taux de faux positifs est \(\approx (1 - e^{-k M / N})^k\). En augmentant \(k\) et/ou \(N\) on réduit ce taux au prix d’une consommation mémoire plus élevée. La garantie d’absence de faux négatifs est particulièrement précieuse dans les applications de filtrage où manquer une valeur interdite a des conséquences graves (détection de contenu illicite, listes noires, etc.).
Plutôt que traiter toutes les données accumulées depuis le début, il est souvent souhaitable de se limiter aux données d’une fenêtre temporelle glissante. Deux raisons principales à cela :
Le volume de données accumulé peut rendre le coût du traitement de toutes les données excessif, même avec un échantillonnage.
Les distributions des attributs sont souvent non stationnaires : remonter loin dans l’historique peut être non pertinent (par ex., pour estimer la popularité actuelle d’un film sorti en salles il y a plusieurs semaines, cumuler le nombre d’entrées par jour depuis la sortie n’est pas utile).
Si \(l\) est la longueur de la fenêtre, les données des tranches \(i, i-1, \ldots, i-l+1\) sont prises en compte et les données précédentes ignorées.
Fig. 105 Fenêtre temporelle glissante appliquée à un flux de données¶
La fenêtre peut être de longueur fixe (cas le plus simple, adapté quand c’est la non-stationnarité qui motive l’emploi de fenêtres) ou de volume fixe (préférable quand c’est le trop grand volume de données qui justifie l’utilisation de fenêtres).
Les fenêtres avec oubli (decaying windows) permettent de pondérer les tranches plutôt que de les inclure ou exclure nettement : la pondération d’une tranche diminue avec l’écart temporel à la tranche courante. Un cas fréquent est la décroissance exponentielle : en notant \(m_i\) la valeur numérique de la tranche \(i\) (par ex. nombre de messages par heure sur un sujet), la mesure de popularité courante \(s_i\) vérifie
\[s_i = m_i + s_{i-1} \cdot d, \ \ s_0 = 0\]
où \(0 < d < 1\) est le facteur d’oubli (decay factor). Cette relation de récurrence évite de conserver les données depuis le début du flux tout en prenant en compte l’historique entier (avec des pondérations décroissantes). Elle permet également de lisser les estimations.
Mise à jour et application de modèles : l’exemple de Streaming K-means¶
Lorsqu’elle est faite à partir de données en flux, la construction d’un modèle doit être incrémentale : il faut pouvoir mettre à jour le modèle existant avec les données d’une tranche sans avoir à le ré-estimer complètement (sur la totalité des données). Par ailleurs, si la distribution des données évolue dans le temps, il peut être utile de donner des pondérations plus fortes aux données récentes, ce qui se traduit par une sous-pondération du modèle antérieur dans la construction du nouveau modèle.
Considérons la classification automatique avec k-means des données d’un flux. Pour k-means, un modèle consiste en l’ensemble des centres de gravité des \(k\) groupes \(\mathcal{C} = \{\mathbf{m}_j, 1 \leq j \leq k\}\). L’algorithme Streaming K-means procède comme suit :
Entrées : tranches du flux, ensembles \(\mathcal{E}_t\) de \(n_t\) données de \(\mathbb{R}^p\), pour \(t = 1, 2, \ldots\) ;
Sorties : après chaque tranche, \(k\) groupes et ensemble \(\mathcal{C_t}\) de leurs centres ;
Initialisation des centres \(\mathbf{m}_{j,1}\), \(1 \leq j \leq k\), à partir des données de la première tranche (par exemple, avec k-means||) ;
pour (chaque tranche \(\mathcal{E}_t\) du flux) faire
Affectation de chaque donnée de la tranche au groupe du centre le plus proche ;
Mise à jour des centres et remplacement des anciens centres par les nouveaux ;
finpour
La mise à jour des centres (étape 2.2) se fait uniquement avec les données de la dernière tranche tout en tenant compte des données plus anciennes à travers les anciens centres. La relation de mise à jour est
où \(n_{j,t}\) est le nombre de données de la tranche courante affectées au groupe \(j\), \(N_{j,t}\) est le nombre total de données affectées à ce groupe depuis le début, \(\mathbf{x}_{j,t}\) est le centre calculé uniquement sur les données de la tranche courante et \(0 \leq \alpha \leq 1\) est le facteur d’oubli. Si \(\alpha = 1\) toute donnée ancienne a autant de poids qu’une donnée récente ; si \(\alpha = 0\), les anciens centres n’interviennent que dans l’affectation initiale des nouvelles données (étape 2.1).
Les illustrations suivantes montrent un exemple simple de classification avec Streaming K-means sur un flux de données de \(\mathbb{R}^3\), généré à partir de sept lois de faible variance (\(\alpha = 0{,}5\), distribution stationnaire).
Fig. 106 Groupes après l’arrivée des premières données¶
Fig. 107 Groupes après l’arrivée de plusieurs tranches de données¶
Spark Structured Streaming est l’API recommandée de Spark pour le traitement de flux de données. Contrairement à l’ancienne API Spark Streaming (basée sur des DStream qui sont des séquences de RDD), Structured Streaming repose sur l’API DataFrame / Dataset, ce qui permet d’écrire des traitements de flux avec la même syntaxe que les traitements par lots (batch).
Le modèle de programmation repose sur l’abstraction d’une table d’entrée infinie : chaque tranche de données arrivant dans le flux est considérée comme une nouvelle ligne ajoutée à cette table. Une requête continue est exprimée comme une transformation de cette table (filtrages, agrégations, jointures, etc.), et le résultat est maintenu à jour dans une table de résultats. L’écriture des résultats peut être faite selon différents modes (append, complete, update).
Fig. 108 Le modèle de la table infinie dans Structured Streaming (figure issue de la documentation de Spark)¶
Les sources de données supportées nativement incluent les fichiers (formats JSON, CSV, Parquet, etc.), les sockets TCP, les rate sources (génération de données synthétiques pour les tests), et Apache Kafka. Les sinks (destinations d’écriture) incluent la console, les fichiers et Kafka.
Spark Streaming (l’ancienne API DStream, basée sur les RDD) reste disponible mais n’est plus activement développée. Structured Streaming est l’API à utiliser pour tout nouveau développement.
Apache Kafka est une plateforme distribuée de streaming d’événements (event streaming platform), créée à l’origine par LinkedIn et devenue un projet de la fondation Apache en 2011. Elle est aujourd’hui l’un des systèmes de streaming les plus utilisés pour la construction de pipelines de données en temps réel.
Par événement (event ou record ou message) on entend ici l’enregistrement de quelque chose qui a eu lieu. Chaque événement est composé d’une clé, une valeur une étiquette temporelle et des métadonnées optionnelles. Par exemple, {key:"Alice",value:"madeapaymentof$200toBob",timestamp:"June25,2026,2:06p.m."}.
Fig. 109 Architecture simplifiée d’Apache Kafka : producteurs, courtiers (brokers) et consommateurs (source Ravi.flipsyde, CC BY-SA 4.0, via Wikimedia Commons)¶
L’architecture de Kafka s’articule autour de quelques concepts fondamentaux :
Producteurs (producers) : les applications qui publient (écrivent) des événements dans Kafka.
Topics : les catégories ou flux d’événements dans lesquels les messages sont publiés. Chaque topic est un journal (log) ordonné et durable des messages. Les topics sont partitionnés : chaque topic est divisé en un ou plusieurs fragments (partitions) qui peuvent être répartis sur plusieurs serveurs (brokers), ce qui permet le parallélisme et la scalabilité.
Courtiers (brokers) : les serveurs Kafka qui stockent les événements et servent les requêtes des producteurs et des consommateurs. Un cluster Kafka est composé de plusieurs courtiers pour assurer la haute disponibilité (via une réplication des partitions).
Consommateurs (consumers) : les applications qui lisent les événements depuis les topics. Les consommateurs peuvent être regroupés en groupes de consommateurs (consumer groups) : chaque partition n’est lue que par un seul consommateur du groupe, ce qui permet de paralléliser la consommation d’un topic.
Fig. 110 Topics partitionnés dans Kafka : chaque partition est un journal ordonné, répliqué sur plusieurs courtiers. Dans l’exemple, le topic a quatre partitions. Deux producteurs différents publient dans ce topic. Les événements avec une même clé (la couleur ici) sont écrits dans une même partition. Source : documentation Kafka¶
Kafka offre plusieurs propriétés qui le distinguent d’une simple file de messages :
Durabilité et rétention configurable. Les messages sont stockés sur disque et conservés pendant une durée configurable (par ex. 7 jours), indépendamment du fait qu’ils aient été consommés ou non. La simple consommation ne les efface pas. Des consommateurs peuvent donc rejouer des événements passés.
Ordre garanti au sein d’une partition. Les messages d’une même partition sont lus dans l’ordre dans lequel ils ont été écrits. L’ordre global sur l’ensemble d’un topic n’est en revanche pas garanti (sauf avec une seule partition). Comme tous les messages avec une même clé sont stockés sur une même partition, l’ordre est garanti entre les messages de même clé.
Scalabilité horizontale. L’ajout de partitions et de courtiers permet d’augmenter le débit du système.
Découplage producteurs / consommateurs. Les producteurs ne connaissent pas les consommateurs et vice versa ; Kafka joue le rôle de tampon (buffer) durable entre eux. Plusieurs applications consommatrices indépendantes peuvent lire le même topic.
Haut débit. Certains déploiements industriels traitent des millions de messages par seconde.
Agrégation de logs : collecte centralisée des journaux applicatifs de nombreux serveurs.
Pipelines de données en temps réel : transfert de données depuis des systèmes sources (bases de données, APIs, capteurs) vers des systèmes cibles (entrepôts de données, moteurs d’indexation, tableaux de bord).
Détection d’événements en temps réel : détection de fraudes, alertes sur des métriques de production, etc.
Découplage de microservices : communication asynchrone entre services dans une architecture distribuée.
La combinaison Kafka / Spark Structured Streaming est aujourd’hui une des solutions les plus répandues dans le traitement de données en temps réel. Kafka est le collecteur et tampon durable : il reçoit les événements provenant de diverses sources (capteurs, applications, APIs) et les met à disposition de façon fiable. Spark Structured Streaming est le moteur de traitement : il consomme les événements depuis Kafka, applique des transformations (agrégations, jointures, modèles prédictifs, etc.) et écrit les résultats vers une destination (base de données, tableau de bord, autre topic Kafka).
Ce modèle permet d’écrire des traitements de flux qui bénéficient à la fois de la durabilité et du haut débit de Kafka, et de la richesse des transformations de Spark. En particulier, Spark maintient automatiquement le décalage (offset) de lecture dans chaque partition Kafka, ce qui garantit un traitement exactement une fois (exactly-once).
Il est important de noter que Kafka est un système à part entière qui doit être installé et administré séparément de Spark.
|
——————–
