clickhouse-io

Modèles d'Analytique ClickHouse

Modèles spécifiques à ClickHouse pour l'analytique haute performance et l'ingénierie des données.

Quand Activer

• Conception de schémas de tables ClickHouse (sélection du moteur MergeTree)
• Rédaction de requêtes analytiques (agrégations, fonctions de fenêtre, jointures)
• Optimisation des performances des requêtes (élagage de partition, projections, vues matérialisées)
• Ingestion de grands volumes de données (insertions par lots, intégration Kafka)
• Migration de PostgreSQL/MySQL vers ClickHouse pour l'analytique
• Mise en œuvre de tableaux de bord en temps réel ou d'analytique de séries temporelles

Vue d'Ensemble

ClickHouse est un système de gestion de base de données (SGBD) orienté colonnes pour le traitement analytique en ligne (OLAP). Il est optimisé pour les requêtes analytiques rapides sur de grands ensembles de données.

Caractéristiques Clés :

• Stockage orienté colonnes
• Compression des données
• Exécution parallèle des requêtes
• Requêtes distribuées
• Analytique en temps réel

Modèles de Conception de Tables

Moteur MergeTree (Le Plus Courant)

CREATE TABLE analytics_marches (
    date Date,
    id_marche String,
    nom_marche String,
    volume UInt64,
    transactions UInt32,
    traders_uniques UInt32,
    taille_moyenne_transaction Float64,
    cree_le DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, id_marche)
SETTINGS index_granularity = 8192;

ReplacingMergeTree (Déduplication)

-- Pour les données pouvant contenir des doublons (par exemple, provenant de plusieurs sources)
CREATE TABLE evenements_utilisateurs (
    id_evenement String,
    id_utilisateur String,
    type_evenement String,
    horodatage DateTime,
    proprietes String
) ENGINE = ReplacingMergeTree()
PARTITION BY toYYYYMM(horodatage)
ORDER BY (id_utilisateur, id_evenement, horodatage)
PRIMARY KEY (id_utilisateur, id_evenement);

AggregatingMergeTree (Pré-agrégation)

-- Pour maintenir des métriques agrégées
CREATE TABLE stats_marche_horaires (
    heure DateTime,
    id_marche String,
    volume_total AggregateFunction(sum, UInt64),
    transactions_totales AggregateFunction(count, UInt32),
    utilisateurs_uniques AggregateFunction(uniq, String)
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(heure)
ORDER BY (heure, id_marche);

-- Interrogation des données agrégées
SELECT
    heure,
    id_marche,
    sumMerge(volume_total) AS volume,
    countMerge(transactions_totales) AS transactions,
    uniqMerge(utilisateurs_uniques) AS utilisateurs
FROM stats_marche_horaires
WHERE heure >= toStartOfHour(now() - INTERVAL 24 HOUR)
GROUP BY heure, id_marche
ORDER BY heure DESC;

Modèles d'Optimisation de Requêtes

Filtrage Efficace

-- PASS: BON : Utiliser les colonnes indexées en premier
SELECT *
FROM analytics_marches
WHERE date >= '2025-01-01'
  AND id_marche = 'marche-123'
  AND volume > 1000
ORDER BY date DESC
LIMIT 100;

-- FAIL: MAUVAIS : Filtrer sur des colonnes non indexées en premier
SELECT *
FROM analytics_marches
WHERE volume > 1000
  AND nom_marche LIKE '%élection%'
  AND date >= '2025-01-01';

Agrégations

-- PASS: BON : Utiliser les fonctions d'agrégation spécifiques à ClickHouse
SELECT
    toStartOfDay(cree_le) AS jour,
    id_marche,
    sum(volume) AS volume_total,
    count() AS transactions_totales,
    uniq(id_trader) AS traders_uniques,
    avg(taille_transaction) AS taille_moyenne
FROM transactions
WHERE cree_le >= today() - INTERVAL 7 DAY
GROUP BY jour, id_marche
ORDER BY jour DESC, volume_total DESC;

-- PASS: Utiliser quantile pour les percentiles (plus efficace que percentile)
SELECT
    quantile(0.50)(taille_transaction) AS mediane,
    quantile(0.95)(taille_transaction) AS p95,
    quantile(0.99)(taille_transaction) AS p99
FROM transactions
WHERE cree_le >= now() - INTERVAL 1 HOUR;

Fonctions de Fenêtre

-- Calcul des totaux cumulés
SELECT
    date,
    id_marche,
    volume,
    sum(volume) OVER (
        PARTITION BY id_marche
        ORDER BY date
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS volume_cumule
FROM analytics_marches
WHERE date >= today() - INTERVAL 30 DAY
ORDER BY id_marche, date;

Modèles d'Insertion de Données

Insertion par Lots (Recommandée)

import { createClient } from '@clickhouse/client'

const clickhouse = createClient({
  url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
  username: process.env.CLICKHOUSE_USER,
  password: process.env.CLICKHOUSE_PASSWORD
})

// PASS: Insertion par lots (efficace)
async function insererTransactionsParLot(transactions: Transaction[]) {
  await clickhouse.insert({
    table: 'transactions',
    values: transactions.map(t => ({
      id: t.id,
      id_marche: t.id_marche,
      id_utilisateur: t.id_utilisateur,
      montant: t.montant,
      horodatage: t.horodatage.toISOString()
    })),
    format: 'JSONEachRow'
  })
}

// FAIL: Insertions individuelles (lentes)
async function insererTransaction(transaction: Transaction) {
  // Ne pas faire ceci dans une boucle !
  await clickhouse.insert({
    table: 'transactions',
    values: [{
      id: transaction.id,
      id_marche: transaction.id_marche,
      id_utilisateur: transaction.id_utilisateur,
      montant: transaction.montant,
      horodatage: transaction.horodatage.toISOString()
    }],
    format: 'JSONEachRow'
  })
}

Insertion en Continu

// Pour l'ingestion continue de données
import { Readable } from 'node:stream'

async function insererEnContinu(sourceDonnees: AsyncIterable<Record<string, unknown>>) {
  await clickhouse.insert({
    table: 'transactions',
    values: Readable.from(sourceDonnees, { objectMode: true }),
    format: 'JSONEachRow'
  })
}

Vues Matérialisées

Agrégations en Temps Réel

-- Créer une vue matérialisée pour les statistiques horaires
CREATE MATERIALIZED VIEW stats_marche_horaires_mv
TO stats_marche_horaires
AS SELECT
    toStartOfHour(horodatage) AS heure,
    id_marche,
    sumState(montant) AS volume_total,
    countState() AS transactions_totales,
    uniqState(id_utilisateur) AS utilisateurs_uniques
FROM transactions
GROUP BY heure, id_marche;

-- Interroger la vue matérialisée
SELECT
    heure,
    id_marche,
    sumMerge(volume_total) AS volume,
    countMerge(transactions_totales) AS transactions,
    uniqMerge(utilisateurs_uniques) AS utilisateurs
FROM stats_marche_horaires
WHERE heure >= now() - INTERVAL 24 HOUR
GROUP BY heure, id_marche;

Surveillance des Performances

Performances des Requêtes

-- Vérifier les requêtes lentes
SELECT
    query_id,
    user,
    query,
    query_duration_ms,
    read_rows,
    read_bytes,
    memory_usage
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query_duration_ms > 1000
  AND event_time >= now() - INTERVAL 1 HOUR
ORDER BY query_duration_ms DESC
LIMIT 10;

Statistiques des Tables

-- Vérifier la taille des tables
SELECT
    database,
    table,
    formatReadableSize(sum(bytes)) AS taille,
    sum(rows) AS lignes,
    max(modification_time) AS derniere_modification
FROM system.parts
WHERE active
GROUP BY database, table
ORDER BY sum(bytes) DESC;

Requêtes Analytiques Courantes

Analyse de Séries Temporelles

-- Utilisateurs actifs quotidiens
SELECT
    toDate(horodatage) AS date,
    uniq(id_utilisateur) AS utilisateurs_actifs_quotidiens
FROM evenements
WHERE horodatage >= today() - INTERVAL 30 DAY
GROUP BY date
ORDER BY date;

-- Analyse de rétention
SELECT
    date_inscription,
    countIf(jours_depuis_inscription = 0) AS jour_0,
    countIf(jours_depuis_inscription = 1) AS jour_1,
    countIf(jours_depuis_inscription = 7) AS jour_7,
    countIf(jours_depuis_inscription = 30) AS jour_30
FROM (
    SELECT
        id_utilisateur,
        min(toDate(horodatage)) AS date_inscription,
        toDate(horodatage) AS date_activite,
        dateDiff('day', date_inscription, date_activite) AS jours_depuis_inscription
    FROM evenements
    GROUP BY id_utilisateur, date_activite
)
GROUP BY date_inscription
ORDER BY date_inscription DESC;

Analyse d'Entonnoir

-- Entonnoir de conversion
SELECT
    countIf(etape = 'a_vu_marche') AS vu,
    countIf(etape = 'a_clique_transaction') AS clique,
    countIf(etape = 'a_termine_transaction') AS termine,
    round(clique / vu * 100, 2) AS taux_vu_a_clique,
    round(termine / clique * 100, 2) AS taux_clique_a_termine
FROM (
    SELECT
        id_utilisateur,
        id_session,
        type_evenement AS etape
    FROM evenements
    WHERE date_evenement = today()
)
GROUP BY id_session;

Analyse de Cohorte

-- Cohortes d'utilisateurs par mois d'inscription
SELECT
    toStartOfMonth(date_inscription) AS cohorte,
    toStartOfMonth(date_activite) AS mois,
    dateDiff('month', cohorte, mois) AS mois_depuis_inscription,
    count(DISTINCT id_utilisateur) AS utilisateurs_actifs
FROM (
    SELECT
        id_utilisateur,
        min(toDate(horodatage)) OVER (PARTITION BY id_utilisateur) AS date_inscription,
        toDate(horodatage) AS date_activite
    FROM evenements
)
GROUP BY cohorte, mois, mois_depuis_inscription
ORDER BY cohorte, mois_depuis_inscription;

Modèles de Pipeline de Données

Modèle ETL

// Extraction, Transformation, Chargement
async function pipelineETL() {
  // 1. Extraction depuis la source
  const donneesBrutes = await extraireDepuisPostgres()

  // 2. Transformation
  const transformees = donneesBrutes.map(ligne => ({
    date: new Date(ligne.cree_le).toISOString().split('T')[0],
    id_marche: ligne.slug_marche,
    volume: parseFloat(ligne.volume_total),
    transactions: parseInt(ligne.nombre_transactions)
  }))

  // 3. Chargement dans ClickHouse
  await insererLotDansClickHouse(transformees)
}

// Exécution périodique
setInterval(pipelineETL, 60 * 60 * 1000)  // Toutes les heures

Capture de Données Modifiées (CDC)

// Écouter les modifications PostgreSQL et les synchroniser avec ClickHouse
import { Client } from 'pg'

const clientPG = new Client({ connectionString: process.env.DATABASE_URL })

clientPG.query('LISTEN mises_a_jour_marche')

clientPG.on('notification', async (msg) => {
  const miseAJour = JSON.parse(msg.payload)

  await clickhouse.insert({
    table: 'mises_a_jour_marche',
    values: [
      {
        id_marche: miseAJour.id,
        type_evenement: miseAJour.operation,  // INSERT, UPDATE, DELETE
        horodatage: new Date(),
        donnees: JSON.stringify(miseAJour.nouvelles_donnees)
      }
    ],
    format: 'JSONEachRow'
  })
})

Bonnes Pratiques

1. Stratégie de Partitionnement

• Partitionner par temps (généralement mois ou jour)
• Éviter trop de partitions (impact sur les performances)
• Utiliser le type DATE pour la clé de partition

2. Clé de Tri

• Mettre les colonnes les plus fréquemment filtrées en premier
• Considérer la cardinalité (haute cardinalité en premier)
• L'ordre impacte la compression

3. Types de Données

• Utiliser le plus petit type approprié (UInt32 vs UInt64)
• Utiliser LowCardinality pour les chaînes répétées
• Utiliser Enum pour les données catégorielles

4. À Éviter

• SELECT * (spécifier les colonnes)
• FINAL (fusionner les données avant la requête à la place)
• Trop de JOIN (dénormaliser pour l'analytique)
• Petites insertions fréquentes (préférer les lots)

5. Surveillance

• Suivre les performances des requêtes
• Surveiller l'utilisation du disque
• Vérifier les opérations de fusion
• Examiner le journal des requêtes lentes

Rappel : ClickHouse excelle dans les charges de travail analytiques. Concevez les tables en fonction de vos modèles de requêtes, insérez par lots et exploitez les vues matérialisées pour les agrégations en temps réel.