De BigQuery à Lakehouse : Comment nous avons construit une plateforme d'analyse de données à l'échelle du pétaoctet - Partie 1

À TRM Labs, nous fournissons des outils d'blockchain intelligence pour aider les institutions financières, les entreprises de crypto-monnaie et les agences gouvernementales à détecter et à enquêter sur la criminalité financière et la fraude liées à la crypto-monnaie.

Notre plateforme analytique traite des pétaoctets de données de blockchain à travers plus de 30 blockchains et répond à plus de 500 requêtes de clients par minute avec une latence ultra-faible, alimentée par Postgres et BigQuery distribués. Pendant des années, nous avons optimisé BigQuery pour qu'il évolue efficacement, mais lorsque nous avons eu besoin de l'exécuter sur site, nous nous sommes heurtés à un mur. BigQuery ne pouvait pas prendre en charge nos besoins multi-environnements, et la mise à l'échelle de Postgres pour l'ingestion massive de données devenait trop coûteuse. Nous avions besoin d'une solution ouverte, auto-hébergée, sécurisée et performante. C'est pourquoi nous avons construit un lac de données à l'échelle du pétaoctet avec Apache Iceberg et StarRocks pour répondre à nos besoins d'analyse orientés utilisateur. Voici comment nous avons procédé, ce que nous avons appris et pourquoi cela pourrait changer votre façon d'envisager l'architecture des données.

1. Plate-forme de données de première génération

Dans notre plateforme de données de première génération, nous avons utilisé un regroupement Postgres distribué(Citus Data) pour les recherches rapides de points et les petits pré-agrégats. Lorsque les charges de travail dépassaient la capacité du regroupement Postgres distribué, nous fédérions les requêtes plus importantes et les agrégations ad hoc vers BigQuery.

2. Au-delà de BigQuery et vers un centre de données ouvert de nouvelle génération

Alors que BigQuery a bien répondu aux besoins d'analyse de nos clients pendant des années, nous avons rencontré des limites lorsque nous nous sommes développés dans des déploiements multi-environnements, y compris des environnements sur site. La nécessité de partager les données d'Analyse Blockchain sur plusieurs sites a rendu les services gérés comme BigQuery peu pratiques, et nos charges de travail de service ont nécessité une nouvelle approche de mise à l'échelle.

Exigences clés à l'origine du changement

• Déploiement multi-sites: La nécessité de déployer notre plateforme sur plusieurs sites sur site tout en maintenant les capacités de partage des données a nécessité l'utilisation de solutions open source pouvant être déployées sur Kubernetes.
• Échelle et performances: Notre plus grande charge de travail en contact avec la clientèle, qui s'appuyait auparavant sur BigQuery, contient plus de 115 To de données et croît de 2 à 3 % par mois. Ces requêtes de lecture impliquent des jointures complexes à plusieurs niveaux avec des filtres basés sur le temps et les tableaux. Respecter une latence P95 de trois secondes à un niveau de concurrence élevé avec BigQuery s'est avéré difficile sans investir dans des créneaux de calcul coûteux. Le transfert de charges de travail de ce type vers Postgres distribué serait coûteux, ne serait-ce que pour des raisons de stockage.

Notre plate-forme de données de nouvelle génération devait combiner la flexibilité d'un lac de données avec les performances et la fiabilité d'un entrepôt de données. La construction d'un lac de données moderne autour d'Apache Iceberg a permis l'interopérabilité avec les moteurs de requêtes et les moteurs de calcul distribués qui supportent la spécification Iceberg. Après avoir comparé plusieurs moteurs de requête, nous avons choisi StarRocks. La combinaison d'Apache Iceberg et de StarRocks a répondu à nos exigences en matière de déploiement multisite et de performances, tout en offrant des avantages clés pour la croissance future.

Les opportunités que nous avons vues

• Normes ouvertes: L'implémentation open source d'Apache Iceberg permet l'évolution des schémas, le voyage dans le temps et la gestion efficace des métadonnées sur le stockage d'objets. Sa flexibilité permet un déploiement dans des environnements multi-sites sur site, ce qui le rend parfait pour partager les données d'Analyse Blockchain sur plusieurs sites.
• Lac de données haute performance: StarRocks offre une latence ultra-faible et une concurrence élevée grâce à une mise en cache avancée et à un traitement des requêtes entièrement vectorisé. La combinaison de StarRocks et d'Iceberg nous permet d'obtenir des performances d'entrepôt de données tout en conservant la flexibilité du lac de données.
• Indépendance du moteur de requête: La construction de notre lac de données sur Apache Iceberg nous donne la flexibilité d'intégrer n'importe quel moteur de requête compatible, ce qui nous permet de nous adapter à l'évolution de la technologie. Au cours de l'année qui s'est écoulée depuis la réalisation de nos benchmarks, nous avons observé des progrès rapides dans les performances des moteurs de requêtes. Nous sommes impatients de réévaluer les solutions existantes (par exemple, Trino et DuckDB) et les nouveaux venus (par exemple, Clickhouse et Crunchy Data Warehouse). Cette flexibilité nous permet de rester à la pointe de la performance et de la rentabilité, indépendamment de tout fournisseur.
• Réduction des coûts : Les données et les métadonnées étant stockées efficacement sur le stockage objet, nous avons identifié une opportunité de migrer les charges de travail de notre regroupement Postgres distribué, réduisant ainsi les coûts de stockage SSD.

3. Pourquoi Apache Iceberg + StarRocks pour un lac de données ?

Les déploiements multi-environnements, y compris sur site, devenant une exigence clé, nous avions besoin d'une solution alternative pour les clients confrontés à des cas d'utilisation analytiques. Notre travail avec BigQuery et Postgres nous a permis de faire quelques observations clés :

1. Il est essentiel de minimiser les données lues au moment de la requête en utilisant la compression des données, le regroupement et le partitionnement pour optimiser les balayages.
2. Les index traditionnels de type B-tree deviennent inefficaces à l'échelle du pétaoctet.
3. L'exécution vectorielle moderne de l'unité centrale (par exemple, SIMD) accélère considérablement le traitement des requêtes.
4. La mise à l'échelle horizontale permet une forte concurrence tout en maintenant les coûts à un niveau raisonnable.
5. La séparation du calcul et du stockage nous permet de passer en souplesse d'un moteur de requête à l'autre ou de les combiner pour optimiser les performances de la charge de travail, sans dupliquer les données.

Sur la base de ces informations, nous sommes allés au-delà des magasins de données OLAP traditionnels (par exemple, Clickhouse) et avons commencé à explorer le marché émergent des "entrepôts de données" (Data Lakehouse). Nous devions prendre deux décisions clés : (1) notre format de stockage et (2) notre moteur de recherche.

3.1 Format de stockage

Chez TRM, nos besoins en stockage, en particulier avec l'avènement des blockchains à haut débit, augmentent de façon exponentielle chaque année. Nous devions nous assurer que notre système de stockage était à la fois performant et rentable, car nous allions intégrer d'autres blockchains à l'avenir.

En commençant par le coût, nous savions que nous devions abandonner les disques SSD au profit des magasins d'objets, car même le magasin d'objets le plus cher est quatre fois moins cher que le disque SSD le moins cher.

Notre choix s'étant porté sur les magasins d'objets, nous avons évalué trois des formats de stockage les plus populaires pour la construction d'un lac de données.

Bien que Delta Lake offre des fonctionnalités et des performances intéressantes, nous l'avons écarté en raison de son manque d'évolution des partitions et de son chevauchement avec Iceberg pour les analyses à grande échelle et le traitement par lots. Nous avons ensuite comparé Apache Hudi ; notre table Hudi la plus performante était trois fois plus lente qu'Apache Iceberg.

Nous avons parié sur Apache Iceberg, qui offrait des performances de lecture exceptionnelles tout en bénéficiant d'une large adoption par la communauté, d'une communauté de développement active et d'une prise en charge étendue du catalogue et du moteur de requêtes.

3.2 Moteur de recherche

Après avoir choisi notre format de table, nous avons évalué plusieurs moteurs de requête compatibles avec Iceberg dans leurs versions open source. Nous avons évalué trois moteurs : Trino, StarRocks et DuckDB. Nos analyses comparatives ont montré que StarRocks était systématiquement plus performant que les autres (voir la figure 2 ci-dessous).

• Trino: Un moteur de requête distribué et open source conçu pour interroger de très grands ensembles de données.
• StarRocks: Un moteur de requête rapide et open source pour l'analyse sur et en dehors du lac de données.
• DuckDB: Moteur de requête SQL analytique en cours de traitement, open source.

3.3 Résultats des expériences

Notre expérimentation s'est concentrée sur deux charges de travail principales [6.1.2] : les requêtes de recherche ponctuelle avec filtrage et les requêtes d'agrégation complexes avec filtrage. Nous avons utilisé JMeter pour effectuer des tests de charge et vérifier que les moteurs de requête pouvaient maintenir leurs performances en cas de forte concurrence.

3.3.1 Expérimentation avec Lookup/Filter

La figure 2 montre nos résultats pour cette charge de travail, où nous avons testé des requêtes de consultation de points et de plages qui renvoient de petits sous-ensembles d'un ensemble de données de 2,57 To. Nous avons observé :

• StarRocks: A constamment fourni les meilleures performances dans toutes les configurations, atteignant des temps de réponse aux requêtes aussi bas que 470 ms avec la mise en cache des données.
• Trino: Temps de réponse compris entre 1 410 ms et 1 030 ms, variant en fonction de la taille du regroupement .
• DuckDB: performances raisonnables de 2 à 3 secondes sur un nœud unique puissant. Nous avons interrompu les tests de DuckDB après ce benchmark en raison des limitations de la prise en charge des tables Iceberg. Nous attendons que l'extension Iceberg de DuckDB ajoute la prise en charge du pushdown des prédicats pour une évaluation future.

3.3.2 Expérimentation de l'agrégation complexe

Dans notre expérience suivante, nous avons testé des requêtes qui effectuaient des opérations SUM, COUNT et GROUP BY avec des filtres de tableau et de plage de dates sur un ensemble de données de 2,85 To. Nos résultats sont les suivants :

• StarRocks: StarRocks a géré des charges de travail agrégées complexes de manière exceptionnelle, obtenant des temps de latence d'environ 2 secondes sans mise en cache et aussi bas que 500 ms avec mise en cache sur notre plus grand regroupement test.
• Trino: Bien que les performances de Trino se soient améliorées de manière significative avec des clusters plus importants, elles ont atteint un plafond à environ 2,5 secondes.

3.3.3 Tests de résistance

Nous avons utilisé JMeter pour tester les performances de Trino et de StarRock en cas de forte concurrence.

• StarRocks: StarRocks a toujours été plus performant que Trino lors des tests à haute fréquence pour les charges de travail de consultation et d'agrégation. Lorsque la mise en cache des données a été activée, les performances se sont encore améliorées.
• Trino: Les performances de Trino se sont dégradées à mesure que la charge des utilisateurs simultanés augmentait. Lorsque nous avons effectué ces tests au début de l'année 2024, Trino ne disposait pas de capacités de mise en cache des données pour les tables du lac de données. Bien que cette fonctionnalité ait été ajoutée plus tard dans Trino 439, nous ne l'avons pas encore évaluée.

4. Notre voie à suivre

Sur la base de notre évaluation de trois formats de table ouverts et de l'expérimentation de plusieurs moteurs de requête, nous avons décidé de construire un lac de données avec StarRocks et Apache Iceberg comme composants de base pour répondre aux exigences clés de la construction de la plate-forme de données de TRM sur plusieurs sites et pour améliorer les performances pour nos clients.

• Le data lakehouse offre un double avantage : Notre approche du data lakehouse combine la flexibilité d'un data lake avec la performance d'un data warehouse, ce qui permet des analyses rapides et fiables pour les clients.
• Apache Iceberg : Grâce à ses normes ouvertes, à l'évolution robuste de ses schémas et à la gestion efficace des métadonnées, Iceberg offre l'interopérabilité entre les moteurs dont nous avons besoin.
• StarRocks : Grâce à l'optimisation stratégique du partitionnement des tables Iceberg, du clustering, du dimensionnement du regroupement StarRocks et des stratégies de mise en cache, nous avons obtenu des performances exceptionnelles avec une faible latence et une forte simultanéité. Ces améliorations ont permis d'améliorer de 50 % les temps de réponse P95 et de réduire de 54 % le nombre d'erreurs de dépassement de délai, ce qui nous a permis d'atteindre nos objectifs en matière de performances des requêtes.
• Les tests sont essentiels : Les charges de travail réelles ont révélé des schémas d'utilisation et des possibilités d'optimisation que les tests de référence seuls ne pouvaient pas identifier, soulignant ainsi l'importance cruciale de tests approfondis à grande échelle.

Dans la deuxième partie de cette série, nous examinerons comment nous avons donné vie à cette architecture, du déploiement d'Apache Iceberg sur le stockage objet à l'optimisation de StarRocks pour les déploiements multi-environnements, y compris les environnements sur site.

5. L'ingénierie des données chez TRM

À TRM Labsnous sommes animés par une mission audacieuse : protéger la civilisation contre la criminalité liée à l'IA et construire un monde plus sûr pour des milliards de personnes. En faisant progresser l'blockchain intelligence et en créant la plateforme de données blockchain du futur, nous nous attaquons aux défis les plus difficiles de la criminalité financière et Analyse Blockchain.

Notre mission s'appuie sur des experts tels que :

• Vijay Shekhawat (co-auteur), un membre clé de l'équipe des TRM Labs Data Platform, apporte une expertise approfondie en matière de streaming en temps réel, d' architectures Data Lakehouse et de construction de pipelines sécurisés à haut débit pour l'analyse à l'échelle du pétaoctet, faisant ainsi avancer la mission de TRM.
• Andrew Fisher (co-auteur), ingénieur logiciel au sein de l'équipe des TRM Labs Data Platform, spécialisé dans les chargements de données par lots à grande échelle et les solutions Data Lakehouse qui alimentent les analyses à l'échelle du pétaoctet dans la lutte contre la fraude à la crypto-monnaie.
• Elena Tomlinson, Moamen Ali, Brice Kamgne, Steven Hope, et Sharad Bhadouria, contributeurs clés de l'équipe de recherche de TRM Labs. TRM Labs Data Product Team, ont joué un rôle déterminant dans la validation de l'architecture du lac de données en migrant des charges de travail clés. Leur travail a été essentiel pour garantir l'évolutivité, l'efficacité et la haute performance de la plateforme dans des cas d'utilisation réels.

Remerciements particuliers à Michael Andrews et à Amit Plaha pour leurs commentaires perspicaces et leur dévouement constant à l'excellence technique tout au long de ce projet.

Rejoindre notre équipe

Nos ingénieurs construisent un lac de données à l'échelle du pétaoctet avec des temps de réponse à latence ultra-faible, s'attaquant à certains des défis les plus difficiles d'Analyse Blockchain pour lutter contre la criminalité et construire un monde plus sûr pour des milliards de personnes. Vous êtes enthousiaste et prêt à avoir un impact ? Ou vous pensez pouvoir faire mieux ? Explorez les opportunités et postulez dès aujourd'hui.

‍

{{horizontal-line}}

6. Annexe / expériences détaillées

6.1 - Parcours d'expérimentation

Pour évaluer les performances d'Apache Iceberg en matière de requêtes, nous avons comparé des charges de travail de lecture typiques en utilisant différents moteurs de requêtes afin de trouver celui qui répondait le mieux à nos exigences en matière de performances et d'échelle. Nous nous sommes concentrés sur deux catégories de charges de travail : la recherche/filtre et l'agrégation complexe.

6.1.1 - Préparation des données

‍Pourchaque charge de travail, nous avons créé des tables Iceberg à partir de nos données BigQuery existantes. Nos requêtes font généralement référence à des adresses de blockchain spécifiques, à des entités hors chaîne ou à des périodes de temps, c'est pourquoi nous avons stratégiquement partitionné nos ensembles de données pour nous assurer que les requêtes n'accèdent qu'à des sous-ensembles de données pertinents. En outre :

• Nous avons exporté plusieurs tables (2 à 3 To) de BigQuery au format Parquet sur Google Cloud Storage.
• À l'aide de PySpark, nous avons transformé ces fichiers Parquet en tables Iceberg, en les optimisant à l'aide de configurations de mise en bacs et de tri.
• Nous avons utilisé Dataproc Metastore pour maintenir les métadonnées des schémas et des tables.

6.1.2 - Description de la charge de travail

6.1.2.1 Expérimenter la recherche/filtre

--  Exemple de table Iceberg pour les transactions de la blockchain
CREATE TABLE blockchain_transactions (
    transaction_id STRING,
    numéro_bloc BIGINT,
    from_address STRING,
    to_address STRING,
    amount DECIMAL(38,18), 
    timestamp TIMESTAMP,
    chaîne chaîne
) USING iceberg
PARTITIONED BY (bucket(transaction_id, 3000)) ;

-- Exemple de requête pour filtrer les transactions par ID
SELECT 
    transaction_id,
    from_address,
    to_address,
    montant,
    horodatage
FROM blockchain_transactions 
WHERE transaction_id = '0x1234abcd...';

6.1.2.2 Expérimentation de l'agrégation complexe

--  Exemple de requête pour une agrégation complexe
SELECT
    e.id::TEXT AS entity_id,
    SUM(CASE WHEN t.transaction_type = 'type_a' THEN t.weight_normalized END) AS type_a_volume,
    SUM(CASE WHEN t.transaction_type = 'type_b' THEN t.weight_normalized END) AS type_b_volume, 
    SUM(t.poids_normalisé) AS total_volume,
    SUM(CASE WHEN t.transaction_type = 'type_a' THEN 1 ELSE 0 END) AS type_a_count,
    SUM(CASE WHEN t.transaction_type = 'type_b' THEN 1 ELSE 0 END) AS type_b_count, 
    COUNT(*) AS total_count
FROM transactions t
JOIN entités e ON t.entity_id = e.id
JOIN categories c ON e.category_id = c.id
WHERE t.chain = __chain::TEXT
  AND t.chain_id = __chain_id::TEXT
  AND t.timestamp BETWEEN __start_date::TIMESTAMP AND __end_date::TIMESTAMP
  AND t.path_length BETWEEN __min_hop AND __max_hop
  AND (
        __entity_ids IS NULL 
        OU t.entity_id = ANY(__entity_ids)
      )
  ET (
        __category_ids EST NULL 
        OR EXISTS (
            SELECT 1 
 FROM unnest(t.entity_category_ids) AS cat_id
            WHERE cat_id = ANY(__category_ids)
        )
      )
  AND e.id IS NOT NULL
GROUP BY e.id
ORDER BY __sort_column __sort_order NULLS LAST
LIMIT __limit
OFFSET __offset ;

6.1.3 - Infrastructure

‍Pourévaluer l'évolution des performances, nous avons déployé chaque moteur de requête avec différentes configurations de ressources informatiques. Les spécifications détaillées de la configuration sont fournies dans la section suivante.

• StarRocks: Déployé sur GKE à l'aide de l'opérateur StarRocks Kubernetes.
• Trino: Déployé sur Google Dataproc (plateforme gérée Hadoop/Spark).
• DuckDB: déployé sur des machines virtuelles GCP.

6.2 - Infrastructure

6.2.1 Trino

• Comment: Déploiement sur Google Dataproc (plateforme gérée Hadoop/Spark) pour sa simplicité et ses capacités intégrées de mise à l'échelle automatique.
• Version: Trino 433
• des tailles deregroupement pour l'expérimentation :

6.2.2 StarRocks

• Comment: Déployé sur GKE à l'aide de l'opérateur StarRocks Kubernetes.
• Version: 3.1
• des tailles deregroupement pour l'expérimentation :

6.2.3 DuckDB

• Comment: Déploiement sur les machines virtuelles de Google Cloud Platform (GCP).
• Version: 0.9.2
• des tailles deregroupement pour l'expérimentation :

6.3 - Méthodologie des tests de résistance

Outils et configuration :

• JMeter : Utilisé pour simuler les demandes des utilisateurs au serveur de requêtes.
  • JMeter génère des charges de serveur réalistes en simulant des requêtes d'utilisateurs simultanées. Il prend en charge plusieurs protocoles, notamment JDBC, HTTP et FTP.
  • Composants clés :
    • Groupe de threads : Un ensemble d'utilisateurs simulés effectuant des actions spécifiques, avec un nombre d'utilisateurs, une période de montée en puissance et une fréquence d'exécution réglables.
• Requêtes paramétrées et données CSV : Les requêtes utilisent des paramètres dynamiques provenant de fichiers CSV pour créer des charges de travail variées et réalistes.
• Threads simultanés JMeter : 1, 3, 5, 8, 13, 21, 34, 55 (simule des utilisateurs concurrents)
• Exécution :
  • Durée : Chaque test dure 1 minute en continu.‍
  • Limite d'interrogation : les tests fonctionnent sans limite d'interrogation pour mesurer la capacité maximale du système.