Apache Kafka – Une plateforme centralisée des échanges de données

Kafka est un système de messagerie distribué, originellement développé chez LinkedIn, et maintenu au sein de la fondation Apache depuis 2012.

Son adoption n’a cessé de croitre pour en faire un quasi de-facto standard dans les pipelines de traitement de données actuels.

Bien plus qu’un simple concurrent des outils conçus autour des standards JMS ou AMQP, Kafka a pour ambition de devenir la plateforme centralisée de stockage et d’échange de toutes les toutes les données émises par une entreprise en temps réel.

Dans ce premier article d’une série autour du Projet Apache Kafka, nous verrons la genèse ainsi que les concepts du projet.

La genèse du projet

En 2009 chez LinkedIn, l’équipe en charge de la mise en place de l’adoption d’Hadoop (Jay Kreps, Neha Narkhede, Jun Rao…) s’est vue confrontée à des problèmes quasi insurmontables d’intégration des données provenant des différents systèmes en place au sein de l’infrastructure de l’entreprise.

La grande diversité des technologies utilisées, la fragmentation des données, le couplage fort entre les systèmes induit par les connexions point à point entre ceux-ci, se sont avérés être des freins importants (voir des écueils) pour la création de produits innovants.

Ce couplage fort entre les systèmes producteurs de données et les systèmes consommateurs de données et cette disparité des outils utilisés pour les pipelines de traitements en temps réel, rendait également l’évolution du système très difficile, toute modification d’un outil pouvant entrainer des régressions dans d’autres parties du SI.

 Stream Data Platform

Ce constat a conduit cette équipe à concevoir un outil capable de centraliser les flux de données provenant des différents systèmes d’une entreprise et devant posséder les caractéristiques suivantes :

• Assurer le découplage des systèmes producteurs et consommateurs de données.
• Être un système distribué capable de monter en charge horizontalement (par simple ajout de serveur).
• Être capable de supporter des débits très importants, que ce soit pour la publication ou la lecture de données.
• Supporter la consommation d’un flux de données par des consommateurs multiples.
• Permettre la consommation des messages en temps réel et en mode batch, en persistant les données.

Les outils traditionnels de brokers de messages d’entreprise (RabbitMQ/ActiveMQ) ne se sont pas avérés adaptés à la création de cette plateforme.

Parmi les problèmes rencontrés dans la création d’un tel système avec ActiveMQ/RabbitMQ :

• Le prix prohibitif.
• L’impact négatif sur les performances de l’accumulation de données (pour permettre la consommation de celles-ci par des systèmes batch).
• La persistance optionnelle des données.
• La difficulté pour garantir un ordre de consommation des messages identique à l’ordre de création de ceux-ci (particulièrement dans un système distribué).
• La suppression de la donnée après consommation dans ces systèmes.

C’est ce qui a motivé la création d’un outil ad hoc : Kafka.

Kafka

Kafka est donc un système de messagerie distribué, en mode publish-subscribe, persistant les données qu’il reçoit, conçu pour facilement monter en charge et supporter des débits de données très importants.

Kafka conserve les données qu’il reçoit dans des topics, correspondant à des catégories de données.

On nomme les systèmes qui publient des données dans des topics Kafka des Producers.

Les Consumers, sont les systèmes qui vont lire (transfert de données de type pull) les données des topics Kafka .

Toutes les étapes de ce système peuvent être distribuées :

• De multiples producers peuvent émettre des données vers un même topic.
• Les données d’un topic sont partitionnées et répliquées au sein des différents brokers ou nœud du cluster Kafka.
• Un consumer d’un topic peut lui même être un système distribué (Hadoop, Spark, …).

Log

Le log est l’abstraction de base au cœur du système Kafka.

Il ne s’agit pas ici de logs comme on l’entend lorsque l’on parle des logs d’un serveur Apache ou d’une application Web, mais plutôt d’une structure de données abstraite ayant les caractéristiques suivantes :

• Il s’agit d’un Array de messages.
• Dans lequel les données sont ordonnées selon le temps de réception des messages.
• Chaque nouvelle donnée sera ajoutée à la « fin » du log (append only).
• Les données sont immuables.
• Un index unique, ici nommé l’offset, est attribué à chaque message.

Topic

Pour assurer un débit d’écriture et de lecture très important, les topics seront divisés en partitions distribuées au sein du cluster Kafka. Chaque partition sera répliquée (facteur de réplication configurable par topic) au sein du cluster pour assurer une tolérance à la panne.

Les messages, constitués d’une clé optionnelle et d’une valeur, seront envoyés par les producers à un topic Kafka sous la forme d’un simple tableau de bytes, ce qui permettra de transmettre et de persister des messages dans un format arbitraire (JSON, Avro, …). Les messages possédant une clé commune seront automatiquement envoyés à la même partition d’un topic (la partition étant choisie en effectuant un hash sur la clé d’un message modulo le nombre de partitions du topic).

Chacune des partitions d’un topic se comporte comme un commit log. Contrairement aux autres brokers de messages, Kafka ne va pas maintenir une liste des consommateurs d’un topic ou de leur état de consommation, mais va persister sur disque tous les messages en leur assignant un offset au sein d’une partition d’un topic. La rétention des messages sera configurée selon une durée ou une taille des logs. Charge aux différents consommateurs de maintenir leur état de consommation des différentes partitions d’un topic (l’offset du dernier message consommé). Kafka va ainsi pouvoir supporter un nombre très importants de consommateurs, sans impact sur ses performances.

La consommation des messages d’un topic peut elle même être réalisée par un système distribué (Hadoop, Spark, Akka cluster…). Kafka introduit ainsi la notion de groupe de consommation.

Un consumer group distribue la lecture des partitions d’un topic au sein de ses membres. Chacun des membres d’un groupe de consommation va lire les données d’une ou de plusieurs partitions d’un topic. Une partition d’un topic ne sera consommée à tout moment que par un seul membre d’un groupe de consommation. Le nombre de partitions d’un topic définit ainsi le parallélisme maximal de consommation de ses messages.

Cette consommation des messages sera tolérante à la panne d’un membre d’un groupe de consommation. Lorsque qu’un consumer disparait, les partitions qui étaient consommées par celui-ci se voit assigner un autre membre du groupe de consommation (consumer rebalance).

Au delà d’un simple système de messages, on définit ainsi souvent Kafka comme un système de commit logs distribué (Distributed commit log).

La persistance des messages dans des commit log va permettre à un nombre arbitraire de consommateurs (ou groupe de consommation) d’extraire les messages d’un même topic (dans lequel l’ordre des messages possédant une clé commune sera garanti), chacun à leur propre rythme. Un même topic va donc pouvoir être utilisé comme la source de données d’un système batch, d’un système de stream processing ou de tout autre système ad hoc.

Opérations

Bien qu’il persiste tous les messages sur disque, Kafka va tout de même bénéficier de performances exceptionnelles.

Source : https://softwaremill.com/mqperf/

Dans la pratique, les partitions d’un topic seront matérialisées par un répertoire dans le file system du serveur sur lequel tourne un broker Kafka. Les messages reçus par ce broker vont tout simplement être ajoutés à la fin d’un fichier (log segment) dans le répertoire correspondant à la partition des messages. Kafka va effectuer une « rotation » des fichiers de segment de log au sein d’une partition, créant un nouveau segment dès qu’un fichier atteint une taille prédéfinie (1G par défaut), et supprimant les fichiers les plus anciens (en fonction d’une durée de rétention ou de la taille maximale d’une partition). Il existe également un système de Log Compaction qui permet de ne retenir que la dernière valeur associée à chaque clé des messages d’un topic.

Comment expliquer les bonnes performances d’un système qui persiste tous les messages qu’il reçoit sur disque ?

Tout d’abord, Kafka va faire en sorte qu’un maximum de petites opérations d’écritures provoquent une seule écriture sur disque (batching). De plus, de par la nature des commit logs la majorité des opérations sur disque seront séquentielles ce qui entraine d’excellentes performances en terme de débit de lecture/écriture. Les lectures/écritures séquentielles sur disques sont plus performantes que des lectures/écritures aléatoires en mémoire !

Source : http://queue.acm.org/detail.cfm?id=1563874

De plus, les OS modernes vont utiliser la mémoire disponible comme un cache des opérations sur disque (pagecache) rendant les écritures aléatoires sur disque particulièrement rapides. Ce mécanisme sera de toutes les façons utilisé par les systèmes qui maintiennent eux même un cache des données en mémoire tout en les persistant sur disque, stockant de fait deux fois les données dans la RAM! Autant déléguer l’utilisation de la RAM comme cache des opérations sur disque à l’OS (c’est également le fonctionnement du cache HTTP Varnish).

L’overhead en mémoire des objets dans la JVM peut être très important (La taille des objets pouvant parfois avoir une taille double de la taille des données qu’ils contiennent). En ne conservant pas un cache en mémoire des données on s’affranchit de cet overhead ainsi que de la gestion complexe du Garbage Collector.

Enfin pour envoyer des messages aux consommateurs, Kafka va s’appuyer sur un mécanisme très performant en zero copy (commande linux sendfile). Les données des messages ne transiteront même pas par la couche applicative du broker Kafka, passant « directement » de la tête de lecture sur disque à la socket réseau.

Il est également possible d’utiliser la compression end-to-end de groupes de messages lorsque la bande passante du réseau s’avère être le principal bottleneck.

Version 0.9

L’absence de sécurité dans Kafka fut sans doute un frein à l’adoption de cet outil dans les entreprises. C’est désormais chose réparée avec la version 0.9 de Kafka qui apporte des nouvelles fonctionnalités :

• L’authentification des clients Kafka est assurée à l’aide du protocole SSL ou SASL (Kerberos).
• Le protocole SSL permet également le chiffrement des données entre les brokers et les clients du cluster.
• Un mécanisme d’autorisation des lectures/écritures est ajouté.

On notera également l’arrivée de Kafka Connect, framework permettant la réalisation de connecteurs pour Kafka. Il permet d’abstraire le développeur des problèmes classiques à résoudre lors de la création d’un connecteur pour Kafka : gestion des offsets, partitionnement, tolérance à la panne…

De nombreux connecteurs ont dors et déjà été implémentés pour permettre la transmission des données d’un système (HDFS, Cassandra, ElasticSearch, JDBC, …)  depuis ou vers un cluster Kafka: http://www.confluent.io/developers/connectors

Usage et Adoption

Les cas d’utilisation de Kafka dans un pipeline de traitements de données sont variés :

• Agrégation de logs.
• Stream Processing.
• Monitoring.
• Bus de messages.
• Event Sourcing.
• …

De plus en plus d’entreprises utilisent cet outil dans leur data center (voir Power By).

A noter :

• Plus d’un trillion de messages transite par jour sur le cluster Kafka de LinkedIn! (ce qui implique des pics à 4.5 millions de messages par seconde pour un total de plus de 1.34 PB par semaine!)
• Microsoft utilise également un cluster de plus de 1000 brokers Kafka qui ingère jusqu’à 1 million de messages par seconde et délivre jusqu’à 5 millions de messages par seconde!
• Netflix prévoit une intensification de l’utilisation de Kafka dans son pipeline de traitement de données qui utilise jusqu’à 500 milliards d’événements par jour (soit environ 1.3 PB de données par jour!)
• D’autre entreprises qui utilisent Kafka : Twitter, Spotify, Mozilla, Airbnb, Square, Uber, Criteo, OVH, …

Conclusion

Le design unique de Kafka rend ce projet à même de résoudre les problèmes d’intégration de données au sein d’une entreprise :

• En supportant des débits de données très importants.
• En persistant les données qu’il reçoit ce qui permet à plusieurs systèmes cibles de consommer les données provenant d’une même source chacun à leur rythme.
• En étant capable de monter en charge par simple ajout de serveur.
• Et surtout en découplant les systèmes producteurs de données des systèmes consommateurs de données.

Si Hadoop s’est imposé comme l’outil capable de recueillir toutes les données d’une entreprise pour pouvoir être analysées par des systèmes en mode batch (le Data Lake), Kafka va-t-il s’imposer comme le hub centralisé (Stream Data Platform) de tous les évènements émis en temps réel au sein d’une entreprise?

Dans un prochain article, nous explorerons les APIs Java des Producers et Consumers Kafka.