Chaque jour chez Babbar, nous crawlons plusieurs milliards de pages qu’il nous faut analyser et stocker. Dans le même temps nous calculons les métriques de plus de 250 milliards de pages qu’il nous faut mettre à jour en continu (pour un total de plus de 1500 milliards de pages) et propager dans un graphe de près de 15000 milliards de liens.
Pour rappel, à son lancement, la totalité de l’index de google était de moins de 30 millions de pages, ce qui représente 10 secondes de traitement sur notre infra. Aujourd’hui, nous allons lever le voile sur les technologies qui nous permettent d’indexer et de manipuler une telle quantité de données.
L’index de Babbar
Un index, c’est globalement une grosse base de données distribuée et organisée de telle sorte qu’elle autorise un accès rapide aux informations pour nos clients, que cela concerne les pages, les liens ou les sites. De fait, il y a donc plusieurs index. Cela signifie aussi qu’il y a au moins deux aspects : la distribution des données et leur stockage.
En réalité, il y a un troisième aspect et pas des moindres : nous ne nous contentons pas de stocker des pages ou des liens, nous devons également calculer et mettre à jour en continu des métriques pertinentes issues du graphe du web formé par ces données. Nous avons donc besoin d’insérer et de mettre à jour une énorme quantité de donnée là où la plupart des bases de données sont optimisées pour des accès en lecture.
Nos premiers essais d’architecture pour l’index ont considéré des solutions “tout-en-un” comme Apache Ignite, mais malgré de nombreuses tentatives, les performances n’ont jamais atteint les résultats espérés. Nous avons donc pivoté vers une solution “maison”.
Nous allons voir ensemble les différents volets qui composent l’index de Babbar : le partitionnement, la communication via Apache Pulsar, le stockage avec RocksDB et la mise en jour en continu.
Le partitionnement
Le principe de base de la distribution de donnée est simple : on effectue un partitionnement (sharding) des données selon une clef judicieusement choisie de manière à ce que les partitions soient de taille similaires tout en préservant une information relationnelle entre les données au niveau de la partition.
Par exemple, on peut définir une règle qui, pour une URL donnée, effectue un partitionnement selon le Host de l’URL. Ainsi, les Hosts sont répartis dans différentes partitions mais les URLs d’un même Host sont groupés dans la même partition.
Il est également possible de choisir un partitionnement différent pour une même donnée selon le contexte (index vs crawl par exemple), tant que tout le monde est d’accord sur le schéma de partitionnement.
La distribution – Apache Pulsar
Apache Pulsar est un framework open-source, initialement développé par Yahoo, offrant une solution de communication par message (messaging) groupés par sujets (topics) au sein d’espaces de noms (namespaces) selon un modèle de publication-abonnement (publish-subscribe). C’est un mécanisme fondamentalement asynchrone entre les nœuds connectés à un même cluster Pulsar.
Pour notre usage, les noms des topics au seins de leur namespace sont de simples numéros correspondants aux identifiants des partitions définies par le schéma. Par exemple, babbar.tech peut se trouver dans la partition 42 et yourtext.gurudans la partition 69. Un consommateur peut s’inscrire pour recevoir uniquement les fetches (namespace) de la partition 42 (topic) mais un autre peut n’être intéressé que par les crawl requests (namespace) de la partition 69 (topic).
Un Framework flexible
Pulsar supporte un ensemble de fonctionnalités intéressantes comme :
- la compression des messages – lorsque la quantité de donnée est importante et/ou que les messages sont gros, pouvoir choisir et activer un mécanisme de compression devient rapidement nécessaire
- le traitement par lot (batching) – lorsque les messages sont petits mais très nombreux, il est important de pouvoir les traiter par lot, tant pour l’efficacité de l’algorithme de compression que pour les performances à leur émission et réception
- les souscriptions multiples – il est possible d’avoir plusieurs consommateurs sur un même topic et de choisir entre une duplication des messages pour chaque consommateur ou une ventilation des messages entre les consommateurs
- le basculement (failover) – dans un environnement distribué, il peut être intéressant d’avoir un destinataire alternatif au cas ou le consommateur prévu pour un topic devient inaccessible
- la persistance (persistence) – il peut être nécessaire d’être assuré qu’aucun message n’est perdu via un mécanisme d’acquittement et de redistribution des messages en cas de problème
- réactif vs proactif – chaque consommateur peut décider de traiter les messages via des événements (push) ou d’en faire la demande explicite (pull)
Pulsar permet donc une extensibilité horizontale. Besoin de plus de crawlers ? Il suffit de connecter les machines au cluster et de mettre à jour la configuration des consommateurs de crawl requests !
Parcours d’une URL
Le parcours typique d’une crawl request et du fetch résultant peut se résumer ainsi :
- la politique de crawl place une requête de crawl dans le namespace crawl requests
- un crawler va consommer la requête de crawl et fetcher la page web
- après traitement, le résultat du fetch sera placé par le crawler dans le namespace fetches, dans le topic correspondant au Host tel que définit par le schéma de partitionnement
- l’index va consommer et stocker le fetch mais également analyser les liens sortants et placer des messages dans les topics du namespace links selon leur Host cible
- les index vont consommer et stocker les liens d’un topic particulier correspondant à la partition dont ils ont la charge
Le stockage – RocksDB
RocksDB est une base de donnée clef-valeur (NoSQL) performante et open-source. Elle est issue d’un fork de LevelDB (développé par Google) et a été optimisée pour les processeurs multi-cœurs et les disques flash (SSD) par Meta (anciennement Facebook) et est maintenue par ces derniers.
Principe
RocksDB (comme LevelDB) s’appuie sur une structure de données particulière : les LSM tree (Log-Structured Merge-tree) multi-niveaux. Chaque niveau représente une consolidation du précédent, avec les informations les plus récentes dans les premiers niveaux. Les niveaux sont divisés en fichiers SST (Static Sorted Table) contenant des opérations, triées par la clef de l’entrée sur laquelle elles opèrent.
Les seules opérations supportées sont : put, delete et merge. Dans ce contexte, un delete ne supprime rien, au contraire il ajoute une opération dans le niveau le plus récent, c’est la consolidation qui effectue une réduction sur les opérations présentes dans les niveaux concernés : un put ou un delete récent pour une entrée remplace les opérations plus anciennes sur cette même entrée. On ajoute donc des opérations dans le premier niveau et lorsque sa taille dépasse une certaine limite par rapport aux niveaux suivants, une consolidation est effectuée (compaction) et un niveau existant est re-généré en incluant tous les niveaux au dessus de lui. Ci-après, la situation après la consolidation des deux premiers niveaux :
La consolidation doit aussi être effectuée au vol lors d’une opération de lecture : il faut traverser tous les niveaux, du plus récent au plus ancien, en effectuant la réduction des opérations concernant l’entrée recherchée pour obtenir son état final. Un get peut donc avoir un coût non négligeable.
Les opérateurs de merge
Vous aurez probablement noté qu’il n’existe pas d’opération de modification dans un tel système. Si, par exemple, on souhaite mettre à jour la date de dernière rencontre d’une URL sans modifier ses métriques déjà calculées, il faut une lecture pour récupérer les informations de l’URL, modifier la date en question puis émettre une opération d’écriture avec les informations à jour. Compte tenu du coût d’une opération de lecture, cette solution ne passe pas à l’échelle. C’est la qu’interviennent les opérations de merge.
Le merge permet de combiner deux états (courant et précédent) d’une entrée pour produire un état à jour. Le comportement de cette opération doit être définie par l’utilisateur via une fonction qui sera appelée à chaque fois que nécessaire. En définissant correctement la structure des informations stockées et l’opérateur de merge associé, il est possible d’effectuer une mise à jour des données en la différant à l’étape de consolidation.
Ceci amène une difficulté technique : en cas de consolidation impliquant une opération de merge, c’est la base de données elle-même qui doit faire appel à la fonction fournie par l’utilisateur. Or, RocksDB est implémenté en C++, nos applications sont en Java et l’API de RocksDB ne permet pas de définir nos propres opérateurs en Java. Il a donc fallut se retrousser les manches et intervenir dans le code de RocksDB pour le modifier de manière à supporter la définition d’opérateurs – codés en Java – qui seront appelés pendant l’opération de consolidation. Et cela ne fut que la première étape, car le coût d’un appel C++ vers Java n’est pas négligeable et une consolidation entre deux niveaux les plus profonds peut nécessiter des millions d’appels.
C’est donc l’implémentation d’une version optimisée du support des opérateurs de merge qui fut le véritable game changer dans notre architecture en nous permettant la mises à jour de nos données de manière performante. Tant que nous y étions, nous avons également ajouté le support des filtres à la consolidation (Compaction Filters). Et puisque RocksDB est open-source, vous pouvez trouver notre contribution sur github !
La mise à jour en continu
Une fois le problème de l’ingestion des données et leur mise à jour résolu, il nous fallait mettre en place un moyen de calculer les métriques de graphe du web en continu ainsi que d’autres tâches automatisées.
C’est à ce moment que les différentes pièces du puzzle doivent s’emboîter. Nous avons des centaines de base de données distribuées sur des dizaines de machine et il faut donner du sens à tout cela ! C’est le choix des index, du format des clefs et du schéma de partitionnement qui nous garantissent la co-localisation des informations au moment où nous en avons besoin.
Ainsi nous pouvons définir un parcours synchronisé de plusieurs index présents sur le même disque de la même machine afin de regrouper toutes les informations concernant un même site web et de calculer ses métriques. Grâce au schéma de partitionnement et à Pulsar, nous pouvons diffuser ces métriques par une variante du surfeur raisonnable via les liens sortants du site. Enfin, à réception, un opérateur de merge personnalisé nous permet de mettre à jour les métriques des liens entrants tout en conservant les autres informations en provenance des crawlers.
C’est aussi ce parcours synchronisé et les opérations de merge qui nous permettent de maintenir le graphe du web en temps réel en tenant compte des mises à jour des pages web et des liens sortants qu’elles comportent. Il faut être capable de générer la différence entre deux versions successives d’une même page afin de créer, supprimer ou mettre à jour des backlinks qui se trouvent sur une autre partition et donc sur une autre machine.
Enfin, il est nécessaire d’avoir un ramasse-miettes pour une maintenir une bonne hygiène dans la base de données, et supprimer une page web de l’index nécessite également la suppression des backlinks.
Les fondations de Babbar
C’est en construisant notre propre solution pour l’index que nous avons posé les bases de l’architecture de Babbar. Ce sont ces innovations techniques qui nous permettent de maintenir à jour une image du web à grande échelle en temps réel à des coûts supportables.
En nous appuyant sur ces fondations techniques, nous sommes en mesure de calculer et de fournir des métriques représentatives comme le Babbar Authority Score, la Semantic Value ou la Force Induite. Et nous avons d’autres idées en cours de développement, stay tuned !
