Data Marts de Delhivery - Parcours de migration d'OLTP vers HTAP

En tant que principale plateforme de traitement des commandes pour le commerce numérique en Inde, Delhivery traite un million de colis par jour, 365 jours par an. Ses 24 centres de tri automatisés, 101 hubs, plus de 3 100 centres de livraison directe, plus de 1 000 centres partenaires, plus de 11 000 flottes et plus de 60 000 membres d'équipe fonctionnent sans problème grâce à un vaste réseau d'appareils IoT. Des milliers d’événements et de messages de données entrent et sortent de nos pipelines chaque seconde. Cela équivaut à un énorme volume de données quotidien en téraoctets, ce qui rend la visibilité opérationnelle cruciale pour nous et nos parties prenantes.

Conscients de ces exigences, nous avons décidé de créer des data marts, des bases de données centralisées et finalement cohérentes qui offrent aux utilisateurs un accès rapide aux données commerciales pré-agrégées. Cela permet à nos parties prenantes d'accéder rapidement aux informations commerciales sans avoir à parcourir l'intégralité d'un entrepôt de données.

Cependant, face à une telle ampleur, l'un des défis majeurs consistait à maintenir l'intégrité des données et une faible latence tout en fournissant la capacité nécessaire pour les charges de travail analytiques.

Dans ce blog, je vais dévoiler toutes mes connaissances lors de la migration de nos data marts d'Amazon Aurora vers TiDB, une base de données SQL distribuée de traitement transactionnel/analytique hybride (HTAP). Espérons que cet article puisse fournir des informations aux responsables de l'ingénierie des données, aux administrateurs de bases de données ou aux architectes de données qui envisagent une migration similaire vers TiDB ou toute autre base de données HTAP.

OLTP, OLAP et HTAP

Pour mieux comprendre le cas des data marts en temps réel chez Delhivery, familiarisons-nous d'abord avec trois concepts qui sont au cœur de notre cas d'utilisation : OLTP, OLAP & HTAP :

• OLTP : les systèmes de traitement des transactions en ligne (OLTP) sont conçus pour les applications orientées transactions, garantissant l'intégrité des données grâce aux propriétés ACID (Atomicité, Cohérence, Isolation, Durabilité).
• OLAP : les systèmes de traitement analytique en ligne (OLAP) permettent une analyse multidimensionnelle à grande vitesse de gros volumes de données, facilitant ainsi la prise de décision basée sur les données.
• HTAP : Hybrid Transaction/Analytical Processing (HTAP) combine les fonctionnalités OLTP et OLAP, permettant des analyses en temps réel sur les données transactionnelles.

Cas d'utilisation des datamarts en temps réel chez Delhivery

Les datamarts en temps réel diffèrent des datamarts traditionnels dans la mesure où ils ingèrent des données en temps réel, et non à des intervalles spécifiques. Ces magasins de données sont essentiels à la prise de décision opérationnelle au sol à Delhivery, car nous ne pouvons nous permettre aucun retard dans la synchronisation de ces événements.

Notre parcours vers un datamart en temps réel a commencé en 2020 lorsque nous avons identifié le besoin de tableaux de bord centralisés, en particulier le tableau de bord EYE. L'objectif de ce tableau de bord était de fournir une visibilité opérationnelle en temps réel sur les opérations au sol, permettant une prise de décision basée sur des données actualisées. Voici des exemples d'utilisations :

• Planification et visibilité des véhicules : surveillance en temps réel des horaires de connexion entrants et sortants pour les hubs de Delhivery.
• Suivi des performances : suivi continu des performances des installations de Delhivery.
• Visibilité du contrôle centralisé : fournir à l'équipe centrale des informations précises sur les bloqueurs au sol pour prendre les mesures appropriées. Ces problèmes peuvent être dus à divers facteurs tels qu’une baisse des performances du centre, le vieillissement des expéditions ou la congestion des connexions entrantes et sortantes.
• Conformités : suivi des mesures de conformité put and pick

Mise en œuvre initiale et défis

Nous avons pensé résoudre nos cas d'utilisation à l'aide d'outils d'entrepôt de données tels que Redshift et Snowflake, mais aucune de ces solutions n'a fonctionné pour nous compte tenu du modèle de conception et de l'exigence d'ingestion de données en temps réel ainsi que de fusion.

Ainsi, nous avons initialement choisi Aurora (PostgreSQL) pour servir notre cas d'utilisation de data mart.

Le processus d’ingestion de données autour d’Aurora

Nous avons architecturé nos datamarts en temps réel à l'aide de Spark Streaming et Aurora. Notre pipeline de diffusion était très simple : lire les données de Kafka, traiter les données dans des micro-lots Spark et effectuer des opérations d'insertion dans Aurora.

Notre base de données a été modélisée à l'aide d'une architecture multicouche, composée d'une couche brute, d'une couche partitionnée et d'une couche de data marts. Les utilisateurs n'avaient pas accès pour afficher ou modifier les données dans la couche brute. La couche partitionnée est conservée pour conserver toutes les tables partitionnées (tables de dimension en général). Vous trouverez ci-dessous un schéma simple de notre base de données :

Les défis auxquels nous avons été confrontés avec Aurora

Le système fonctionnait initialement bien, jusqu'à ce qu'il doive gérer un débit supérieur à 3 000 messages par seconde. Cela a marqué le début de plusieurs défis :

• Limitation de l'évolutivité : à mesure que nous dépassions un débit de 3 000 messages par seconde, les limitations des opérations d'entrée/sortie par seconde (IOPS) d'Aurora sont devenues un goulot d'étranglement. La contrainte d’évolutivité avait commencé à impacter nos opérations.

  
  

• Problème de ballonnement : chaque mise à jour d'un enregistrement entraînait la création d'un nouvel enregistrement et d'un tuple mort (version précédente de l'enregistrement). Lorsque le taux de production de ces tuples morts dépassait le processus de nettoyage, des ballonnements se produisaient. Étant donné que VACUUM FULL n'a pas pu réclamer le stockage, l'utilisation du disque a continuellement augmenté. Pour environ 5 To de données, Aurora utilisait plus de 30 To de stockage.

  
  

• Fardeau de maintenance : Le problème des ballonnements est directement lié à nos défis de maintenance. Avec plus de 70 pipelines et un QPS d'écriture total dépassant 5 000 messages/seconde, nous avons constaté que le processus de nettoyage automatique de PostgreSQL, Auto Vacuum, ne parvenait pas à suivre le rythme de génération de tuples morts. Par conséquent, l'exécution manuelle de VACUUM ou VACUUM FULL est nécessaire pour récupérer la base de données. Nos tentatives avec les outils PostgreSQL comme pg_repack et pgcompacttable se sont également révélées infructueuses. Par conséquent, la maintenance est devenue de plus en plus complexe et chronophage.

• Coût : pour répondre à la charge de travail de lecture et d'écriture, nous avons dû évoluer vers les nœuds disponibles les plus élevés (24XLarge). Cela a entraîné une dépense d'environ 100 000 $ par mois pour un cluster Aurora à trois nœuds. À cette échelle, Aurora s'est avéré coûteux en raison de la mise à l'échelle automatique des IOPS.

Recherche d'alternatives

Pour résoudre les limites d'Aurora, nous avons cherché une meilleure alternative répondant aux exigences suivantes :

• Évolutive avec un QPS en écriture élevé : la base de données doit prendre en charge au moins 10 000 QPS en écriture et est évolutive horizontalement.
• Analyse en temps réel : la base de données doit être capable de fournir des fonctionnalités OLAP à haut débit ou en temps réel
• Entièrement distribuée : la base de données doit être distribuée sur plusieurs sites pour offrir une haute disponibilité et une tolérance aux pannes.
• Forte cohérence : la base de données doit maintenir une forte cohérence, garantissant que tous les utilisateurs voient les mêmes données.

Compte tenu de toutes les exigences ci-dessus, nous avons initialement exploré de nombreuses alternatives PostgreSQL, notamment Spanner et Yugabyte, car nous souhaitions minimiser notre gestion des modifications.

Clé

Spanner est un service distribué de gestion et de stockage de bases de données SQL proposé par Google. Il est entièrement géré sur Google Cloud Platform (GCP). Cependant, nous avons constaté que Spanner pourrait ne pas être un bon cas d'utilisation pour notre architecture pour les raisons suivantes :

• Spanner ne prend pas en charge les schémas.
• Nous n'avons pas trouvé les outils appropriés pour charger les données historiques. Nous avons exploré Harbourbridge, un outil open source pour l'évaluation et la migration de Spanner. Cependant, il avait des limites autour de 100 Go de chargement de données.

Yugaoctet

YugabyteDB est une base de données SQL distribuée transactionnelle hautes performances pour les applications cloud natives, développée par Yugabyte. Cette base de données est très proche de notre cas d'utilisation car elle était entièrement conforme à PostgreSQL, évolutive horizontalement et entièrement distribuée. Malheureusement, cela n'a pas fonctionné aussi bien en raison de sa limitation en termes d'évolutivité. Nos critères de réussite exigeaient plus de 7 000 transactions par seconde, mais Yugabyte n'a pu évoluer que jusqu'à 5 000.

Nous avons également examiné d'autres candidats possibles comme BigQuery, mais aucun d'entre eux ne répondait bien à nos exigences.

Atterrissage avec TiDB

Après les alternatives PostgreSQL ci-dessus, nous avons décidé d'ajouter HTAP à nos exigences, ce qui nous a conduit à TiDB. Il prend en charge l'évolutivité, la cohérence, la disponibilité, la topologie de déploiement multi-sites et bien d'autres fonctionnalités prêtes à l'emploi. En tant que base de données distribuée, TiDB comporte plusieurs composants qui communiquent entre eux et forment un système TiDB complet.

• TiDB : il s'agit du composant de traitement SQL sans état qui fournit à l'utilisateur le point de terminaison côté client. Il localise le nœud TiKV correct pour se connecter à partir de PD pour obtenir les données.
• TiKV : il s'agit d'un magasin de données transactionnelles distribuées clé-valeur qui conserve les données dans la plage gauche-fermée-droite-ouverte. Les données sont conservées dans des fragments avec plusieurs réplicas. TiKV utilise le protocole Raft pour la réplication.
• PD : le pilote de placement (PD) conserve les métadonnées du cluster telles que les emplacements des répliques de fragments, et il est également responsable de la planification des fragments sur les nœuds TiKV. Le leader PD gère ces tâches tandis que les autres nœuds maintiennent une haute disponibilité.
• TiFlash : l'extension de stockage en colonnes qui utilise le protocole Multi-Raft Learner pour répliquer les données de TiKV en temps réel, garantissant ainsi la cohérence des données entre le moteur de stockage basé sur les lignes de TiKV.

Les fonctionnalités suivantes de TiDB ont répondu à nos principaux défis et à nos exigences opérationnelles :

• Mise à l'échelle facile

  La conception de l'architecture TiDB sépare l'informatique du stockage, vous permettant d'évoluer ou d'augmenter la capacité de calcul ou de stockage en ligne selon vos besoins. Le processus de mise à l’échelle est transparent pour le personnel d’exploitation et de maintenance des applications.

• Conforme à l'ACIDE

  TiDB est compatible MySQL et prend en charge les transactions prêtes à l'emploi. Il prend en charge les types de transactions optimistes et pessimistes. Cela le rend unique par rapport aux autres bases de données.

• Hautement disponible

  TiKV stocke les données dans plusieurs répliques et utilise le protocole Multi-Raft pour obtenir le journal des transactions. Une transaction ne peut être validée que lorsque les données ont été écrites avec succès dans la majorité des réplicas. Cela garantit une forte cohérence et une haute disponibilité lorsqu’une minorité de répliques tombe en panne.

• HTAP en temps réel

  TiDB combine à la fois le stockage en lignes (TiKV) et le stockage en colonnes (TiFlash) dans la même architecture, formant une pile technologique rationalisée qui facilite la production d'analyses en temps réel sur les données opérationnelles.

Notre infrastructure TiDB

Notre infrastructure TiDB est déployée sur les VM des principaux fournisseurs de services cloud. Nous utilisons TiUP, le gestionnaire de packages de TiDB, pour gérer le cluster et toutes les opérations administratives. Notre cluster est déployé sur 3 zones disponibles (AZ).

Nos configurations de cluster sont les suivantes :

• PD : La couche PD comporte 3 nœuds répartis sur plusieurs AZ. Le leader PD gère ces tâches tandis que les autres nœuds maintiennent une haute disponibilité.
• TiDB : La couche TiDB comporte 9 nœuds de la famille n2-highmem-8. Ces nœuds ont été choisis en fonction des besoins en mémoire, avec 64 Go de RAM et 8 processeurs Core alloués pour chaque nœud TiDB.
• TiKV : La couche TiKV comprend 15 nœuds de la famille n2-highmem-16 qui disposent de 128 Go de RAM et de 16 processeurs vCORE.

En déployant notre cluster TiDB sur plusieurs AZ et en sélectionnant soigneusement les types de nœuds pour répondre à nos besoins de traitement et de mémoire, nous avons créé une infrastructure robuste et hautement disponible, capable de gérer nos exigences de débit de données élevé.

Optimiser TiDB pour notre cas

Pour que cela fonctionne pour notre cas d'utilisation, nous avons travaillé en étroite collaboration avec l'équipe PingCAP pour régler la base de données. Voici quelques-uns des ajustements critiques que nous avons apportés :

Optimisation de l'index

Définissez les paramètres suivants avant de démarrer l'index.

 SET @@global.tidb_ddl_reorg_worker_cnt = 16; SET @@global.tidb_ddl_reorg_batch_size = 4096;

Réinitialisation aux valeurs par défaut après la création de l'index.

 SET @@global.tidb_ddl_reorg_worker_cnt = 4; SET @@global.tidb_ddl_reorg_batch_size = 256;

Élagage de partition

Ceci est principalement important pour les tables partitionnées. Il analyse les conditions de filtre dans les instructions de requête et élimine (élague) les partitions lorsqu'elles ne contiennent aucune donnée requise.

 SET @@session.tidb_partition_prune_mode = 'dynamic';

Analyse de réglage

Parfois, l'analyseur automatique de TiDB échoue si un volume élevé de données est ingéré. Dans ce cas, toutes les requêtes peuvent utiliser le mauvais plan d'exécution et finir par analyser la table entière. Pour éviter une telle situation, nous avons apporté les modifications suivantes aux configurations TiDB :

 set global tidb_max_auto_analyze_time = 86400; set global tidb_enable_pseudo_for_outdated_stats = off; set global tidb_sysproc_scan_concurrency = 15;

Si vous travaillez avec des tables partitionnées, nous vous suggérons d'exécuter manuellement les opérations d'analyse de table pour une partition à la fois afin d'éviter les échecs d'analyse.

Grâce à des ajustements comme ceux-ci, nous avons pu rationaliser efficacement notre utilisation de TiDB, afin d'obtenir des performances optimales pour notre magasin de données en temps réel.

Notre expérience avec TiDB

• Performances améliorées des requêtes

  Nous avons comparé plus de 400 requêtes et constaté que toutes les requêtes s'exécutaient dans le cadre du SLA. Nous avons même constaté un gain de performances de 15 à 20 % pour les requêtes P95.

• Migration facile

  Nous avons utilisé l'outil TiDB Lighting pour migrer toutes les données historiques de notre table de Postgres vers TiDB. Cet outil est très simple à utiliser et très rapide. Nous avons pu charger des téraoctets de données en 2 à 3 heures environ. Cependant, il convient de noter que de nombreux réglages sont nécessaires avant de charger des données aussi volumineuses.

• Un appui solide

  Nous avons rencontré quelques problèmes lors de la configuration de l'infrastructure de production, mais l'équipe d'assistance de PingCAP a joué un rôle très crucial et nous a aidés à ajuster le cluster en fonction de la nature de la charge de travail.

  
  

Conclusion

Dans cet article, nous avons exploré les défis liés à l'utilisation d'Aurora avec notre cas d'utilisation de datamarts en temps réel et le parcours de migration vers TiDB. Nous avons également discuté de la manière dont Delhivery utilise TiDB à grande échelle.

Malgré notre succès avec TiDB, nous reconnaissons qu'aucune solution n'est parfaite et que l'efficacité peut varier en fonction du cas d'utilisation. Dans TiDB, nous avons noté quelques domaines à améliorer, notamment le manque de prise en charge prête à l'emploi pour les vues matérialisées et la gestion native des quotas. Cependant, grâce à des solutions de contournement et des ajustements appropriés, nous avons réussi à remédier efficacement à ces limitations.

Jusqu'à présent, nous avons déployé TiDB dans notre environnement de production. Sur la base de nos benchmarks, TiDB nous permet de traiter plus de milliers de requêtes par seconde avec une latence inférieure à 100 ms. À l’avenir, nous continuerons d’explorer davantage de cas d’utilisation nécessitant une base de données robuste et uniformément distribuée.

Les références

https://docs.pingcap.com/tidb/stable/tidb-lightning-overview

https://reorg.github.io/pg_repack/

https://github.com/dataegret/pgcompacttable

https://cloud.google.com/spanner

https://www.yugabyte.com/yugabytedb/

https://cloud.google.com/bigquery/

https://docs.pingcap.com/tidb/dev/transaction-overview

https://proxysql.com/

Auteur:

Hari Kishan (responsable principal de l'ingénierie @ Delhivery)

Akash Deep Verma (directeur de la technologie @ Delhivery)