Accélérer l’analytique avec MergeTree

Dans la section précédente, vous avez connecté ClickHouse à un catalogue de métadonnées et interrogé directement des formats de table ouverts. Bien qu’il soit pratique d’interroger les données sur place, les formats de table ouverts ne sont pas optimisés pour les charges de travail à faible latence et à forte concurrence qui alimentent les tableaux de bord et le reporting opérationnel. Pour ces cas d’usage, charger les données dans le moteur MergeTree de ClickHouse offre des performances nettement supérieures. MergeTree offre plusieurs avantages par rapport à la lecture directe des formats de table ouverts :

index primaire épars - Ordonne les données sur disque selon une clé choisie, ce qui permet à ClickHouse d’ignorer de larges plages de lignes non pertinentes lors des requêtes.
Types de données enrichis - Prise en charge native de types tels que JSON, LowCardinality et Enum, pour un stockage plus compact et un traitement plus rapide.
Index de saut et index de texte intégral - Structures d’index secondaires qui permettent à ClickHouse d’ignorer les granules ne correspondant pas aux prédicats de filtrage d’une requête, particulièrement efficaces pour les charges de travail de recherche textuelle.
Insertions rapides avec compaction automatique - ClickHouse est conçu pour des insertions à haut débit et fusionne automatiquement les data parts en arrière-plan, de manière analogue à la compaction dans les formats de table ouverts.
Optimisé pour les lectures concurrentes - Le stockage columnar de MergeTree, combiné à plusieurs couches de cache, prend en charge des charges de travail analytiques en temps réel avec une forte concurrence, ce pour quoi les formats de table ouverts n’ont pas été conçus.

Ce guide montre comment charger des données depuis un catalogue dans une table MergeTree à l’aide de INSERT INTO SELECT afin d’accélérer les analyses.

Se connecter au catalogue

Nous utiliserons la même connexion à Unity Catalog que dans le guide précédent, en passant par l’endpoint REST Iceberg :

SET allow_database_iceberg = 1;

CREATE DATABASE unity
ENGINE = DataLakeCatalog('https://<workspace-id>.cloud.databricks.com/api/2.1/unity-catalog/iceberg-rest')
SETTINGS catalog_type = 'rest', catalog_credential = '<client-id>:<client-secret>', warehouse = 'workspace',
oauth_server_uri = 'https://<workspace-id>.cloud.databricks.com/oidc/v1/token', auth_scope = 'all-apis,sql';

Lister les tables

SHOW TABLES FROM unity

┌─name───────────────────────────────────────────────┐
│ unity.logs                                         │
│ unity.single_day_log                               │
└────────────────────────────────────────────────────┘

Explorez le schéma

SHOW CREATE TABLE unity.`icebench.single_day_log`

CREATE TABLE unity.`icebench.single_day_log`
(
    `pull_request_number` Nullable(Int64),
    `commit_sha` Nullable(String),
    `check_start_time` Nullable(DateTime64(6, 'UTC')),
    `check_name` Nullable(String),
    `instance_type` Nullable(String),
    `instance_id` Nullable(String),
    `event_date` Nullable(Date32),
    `event_time` Nullable(DateTime64(6, 'UTC')),
    `event_time_microseconds` Nullable(DateTime64(6, 'UTC')),
    `thread_name` Nullable(String),
    `thread_id` Nullable(Decimal(20, 0)),
    `level` Nullable(String),
    `query_id` Nullable(String),
    `logger_name` Nullable(String),
    `message` Nullable(String),
    `revision` Nullable(Int64),
    `source_file` Nullable(String),
    `source_line` Nullable(Decimal(20, 0)),
    `message_format_string` Nullable(String)
)
ENGINE = Iceberg('s3://...')

Cette table contient ~283 millions de lignes de logs provenant des exécutions de tests CI de ClickHouse — un jeu de données réaliste pour explorer les performances analytiques.

SELECT count()
FROM unity.`icebench.single_day_log`

┌───count()─┐
│ 282634391 │ -- 282.63 million
└───────────┘

1 row in set. Elapsed: 1.265 sec.

Requête sur la table du lac de données

Exécutons une requête qui filtre les logs par nom de thread et type d’instance, recherche des erreurs dans le texte du message et regroupe les résultats par logger :

SELECT
    logger_name,
    count() AS c
FROM icebench.`icebench.single_day_log`
WHERE (thread_name = 'TCPHandler')
    AND (instance_type = 'm6i.4xlarge')
    AND hasToken(message, 'error')
GROUP BY logger_name
ORDER BY c DESC
LIMIT 5

┌─logger_name──────────────┬────c─┐
│ executeQuery             │ 6907 │
│ TCPHandler               │ 4145 │
│ TCP-Session              │  790 │
│ PostgreSQLConnectionPool │  530 │
│ ContextAccess (default)  │  392 │
└──────────────────────────┴──────┘

5 rows in set. Elapsed: 8.921 sec. Processed 282.63 million rows, 5.42 GB (31.68 million rows/s., 607.26 MB/s.)
Peak memory usage: 4.35 GiB.

La requête prend près de 9 secondes car ClickHouse doit effectuer un parcours complet de la table sur l’ensemble des fichiers Parquet du stockage objet. Les performances pourraient être améliorées grâce au partitionnement, mais des colonnes comme logger_name peuvent présenter une cardinalité trop élevée pour permettre un partitionnement efficace. Nous ne disposons pas non plus d’indices tels que les index de texte pour réduire davantage le volume de données à analyser. C’est là que MergeTree excelle.

Charger des données dans une table MergeTree

Créer une table optimisée

Nous créons une table MergeTree avec un schéma optimisé. Notez quelques différences clés par rapport au schéma Iceberg :

Pas d’encapsulation Nullable - supprimer Nullable améliore l’efficacité du stockage et les performances des requêtes.
LowCardinality(String) sur les colonnes level, instance_type, thread_name et check_name - encode par dictionnaire une colonne comportant peu de valeurs distinctes, pour une meilleure compression et un filtrage plus rapide.
Un index de texte intégral sur la colonne message - accélère les recherches textuelles basées sur des tokens, comme hasToken(message, 'error').
Une clé ORDER BY de (instance_type, thread_name, toStartOfMinute(event_time)) - aligne les données sur le disque avec les filtres les plus courants afin que l’index primaire épars puisse ignorer les granules non pertinentes.

SET enable_full_text_index = 1;

CREATE TABLE single_day_log
(
    `pull_request_number` Int64,
    `commit_sha` String,
    `check_start_time` DateTime64(6, 'UTC'),
    `check_name` LowCardinality(String),
    `instance_type` LowCardinality(String),
    `instance_id` String,
    `event_date` Date32,
    `event_time` DateTime64(6, 'UTC'),
    `event_time_microseconds` DateTime64(6, 'UTC'),
    `thread_name` LowCardinality(String),
    `thread_id` Decimal(20, 0),
    `level` LowCardinality(String),
    `query_id` String,
    `logger_name` String,
    `message` String,
    `revision` Int64,
    `source_file` String,
    `source_line` Decimal(20, 0),
    `message_format_string` String,
    INDEX text_idx(message) TYPE text(tokenizer = splitByNonAlpha)
)
ENGINE = MergeTree
ORDER BY (instance_type, thread_name, toStartOfMinute(event_time))

Insérer des données depuis le catalogue

Utilisez INSERT INTO SELECT pour charger les ~300 M de lignes de la table du lac de données dans notre table ClickHouse :

INSERT INTO single_day_log SELECT * FROM icebench.`icebench.single_day_log`

282634391 rows in set. Elapsed: 237.680 sec. Processed 282.63 million rows, 5.42 GB (1.19 million rows/s., 22.79 MB/s.)
Peak memory usage: 18.62 GiB.

Réexécuter la requête

Si nous exécutons maintenant la même requête sur la table MergeTree, nous constatons que les performances s’améliorent considérablement :

SELECT
    logger_name,
    count() AS c
FROM single_day_log
WHERE (thread_name = 'TCPHandler')
    AND (instance_type = 'm6i.4xlarge')
    AND hasToken(message, 'error')
GROUP BY logger_name
ORDER BY c DESC
LIMIT 5

┌─logger_name──────────────┬────c─┐
│ executeQuery             │ 6907 │
│ TCPHandler               │ 4145 │
│ TCP-Session              │  790 │
│ PostgreSQLConnectionPool │  530 │
│ ContextAccess (default)  │  392 │
└──────────────────────────┴──────┘

5 rows in set. Elapsed: 0.220 sec. Processed 13.84 million rows, 2.85 GB (62.97 million rows/s., 12.94 GB/s.)
Peak memory usage: 1.12 GiB.

La même requête s’exécute désormais en 0,22 seconde - soit une accélération d’environ 40x. Deux optimisations clés expliquent cette amélioration :

index primaire épars - La clé ORDER BY (instance_type, thread_name, ...) permet à ClickHouse d’accéder directement aux granules correspondant à instance_type = 'm6i.4xlarge' et thread_name = 'TCPHandler', ce qui réduit le nombre de lignes traitées de 283 millions à seulement 14 millions.
Index de texte intégral - L’index text_idx sur la colonne message permet à hasToken(message, 'error') d’être résolu via l’index plutôt qu’en parcourant chaque chaîne de message, ce qui réduit encore la quantité de données que ClickHouse doit lire.

Le résultat est une requête capable d’alimenter sans difficulté un tableau de bord en temps réel - à une échelle et avec une latence que l’interrogation de fichiers Parquet dans le stockage objet ne peut égaler.

​Se connecter au catalogue

​Lister les tables

​Explorez le schéma

​Requête sur la table du lac de données

​Charger des données dans une table MergeTree

​Créer une table optimisée

​Insérer des données depuis le catalogue

​Réexécuter la requête