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)
• Écriture de requêtes analytiques (agrégations, fonctions de fenêtrage, jointures)
• Optimisation des performances des requêtes (élagage de partitions, 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

Aperçu

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 markets_analytics (
    date Date,
    market_id String,
    market_name String,
    volume UInt64,
    trades UInt32,
    unique_traders UInt32,
    avg_trade_size Float64,
    created_at DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, market_id)
SETTINGS index_granularity = 8192;

ReplacingMergeTree (Déduplication)

-- Pour les données pouvant contenir des doublons (par exemple, provenant de plusieurs sources)
CREATE TABLE user_events (
    event_id String,
    user_id String,
    event_type String,
    timestamp DateTime,
    properties String
) ENGINE = ReplacingMergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (user_id, event_id, timestamp)
PRIMARY KEY (user_id, event_id);

AggregatingMergeTree (Pré-agrégation)

-- Pour maintenir des métriques agrégées
CREATE TABLE market_stats_hourly (
    hour DateTime,
    market_id String,
    total_volume AggregateFunction(sum, UInt64),
    total_trades AggregateFunction(count, UInt32),
    unique_users AggregateFunction(uniq, String)
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, market_id);

-- Interroger les données agrégées
SELECT
    hour,
    market_id,
    sumMerge(total_volume) AS volume,
    countMerge(total_trades) AS trades,
    uniqMerge(unique_users) AS users
FROM market_stats_hourly
WHERE hour >= toStartOfHour(now() - INTERVAL 24 HOUR)
GROUP BY hour, market_id
ORDER BY hour DESC;

Modèles d'optimisation des requêtes

Filtrage efficace

-- RÉUSSI : BON : Utiliser d'abord les colonnes indexées
SELECT *
FROM markets_analytics
WHERE date >= '2025-01-01'
  AND market_id = 'market-123'
  AND volume > 1000
ORDER BY date DESC
LIMIT 100;

-- ÉCHEC : MAUVAIS : Filtrer d'abord sur des colonnes non indexées
SELECT *
FROM markets_analytics
WHERE volume > 1000
  AND market_name LIKE '%election%'
  AND date >= '2025-01-01';

Agrégations

-- RÉUSSI : BON : Utiliser les fonctions d'agrégation spécifiques à ClickHouse
SELECT
    toStartOfDay(created_at) AS day,
    market_id,
    sum(volume) AS total_volume,
    count() AS total_trades,
    uniq(trader_id) AS unique_traders,
    avg(trade_size) AS avg_size
FROM trades
WHERE created_at >= today() - INTERVAL 7 DAY
GROUP BY day, market_id
ORDER BY day DESC, total_volume DESC;

-- RÉUSSI : Utiliser quantile pour les centiles (plus efficace que percentile)
SELECT
    quantile(0.50)(trade_size) AS median,
    quantile(0.95)(trade_size) AS p95,
    quantile(0.99)(trade_size) AS p99
FROM trades
WHERE created_at >= now() - INTERVAL 1 HOUR;

Fonctions de fenêtrage

-- Calculer les cumuls
SELECT
    date,
    market_id,
    volume,
    sum(volume) OVER (
        PARTITION BY market_id
        ORDER BY date
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS cumulative_volume
FROM markets_analytics
WHERE date >= today() - INTERVAL 30 DAY
ORDER BY market_id, 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
})

// RÉUSSI : Insertion par lots (efficace)
async function bulkInsertTrades(trades: Trade[]) {
  await clickhouse.insert({
    table: 'trades',
    values: trades.map(trade => ({
      id: trade.id,
      market_id: trade.market_id,
      user_id: trade.user_id,
      amount: trade.amount,
      timestamp: trade.timestamp.toISOString()
    })),
    format: 'JSONEachRow'
  })
}

// ÉCHEC : Insertions individuelles (lentes)
async function insertTrade(trade: Trade) {
  // Ne pas faire cela dans une boucle !
  await clickhouse.insert({
    table: 'trades',
    values: [{
      id: trade.id,
      market_id: trade.market_id,
      user_id: trade.user_id,
      amount: trade.amount,
      timestamp: trade.timestamp.toISOString()
    }],
    format: 'JSONEachRow'
  })
}

Insertion en continu

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

async function streamInserts(dataSource: AsyncIterable<Record<string, unknown>>) {
  await clickhouse.insert({
    table: 'trades',
    values: Readable.from(dataSource, { 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 market_stats_hourly_mv
TO market_stats_hourly
AS SELECT
    toStartOfHour(timestamp) AS hour,
    market_id,
    sumState(amount) AS total_volume,
    countState() AS total_trades,
    uniqState(user_id) AS unique_users
FROM trades
GROUP BY hour, market_id;

-- Interroger la vue matérialisée
SELECT
    hour,
    market_id,
    sumMerge(total_volume) AS volume,
    countMerge(total_trades) AS trades,
    uniqMerge(unique_users) AS users
FROM market_stats_hourly
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY hour, market_id;

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 size,
    sum(rows) AS rows,
    max(modification_time) AS latest_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(timestamp) AS date,
    uniq(user_id) AS daily_active_users
FROM events
WHERE timestamp >= today() - INTERVAL 30 DAY
GROUP BY date
ORDER BY date;

-- Analyse de rétention
SELECT
    signup_date,
    countIf(days_since_signup = 0) AS day_0,
    countIf(days_since_signup = 1) AS day_1,
    countIf(days_since_signup = 7) AS day_7,
    countIf(days_since_signup = 30) AS day_30
FROM (
    SELECT
        user_id,
        min(toDate(timestamp)) AS signup_date,
        toDate(timestamp) AS activity_date,
        dateDiff('day', signup_date, activity_date) AS days_since_signup
    FROM events
    GROUP BY user_id, activity_date
)
GROUP BY signup_date
ORDER BY signup_date DESC;

Analyse d'entonnoir

-- Entonnoir de conversion
SELECT
    countIf(step = 'viewed_market') AS viewed,
    countIf(step = 'clicked_trade') AS clicked,
    countIf(step = 'completed_trade') AS completed,
    round(clicked / viewed * 100, 2) AS view_to_click_rate,
    round(completed / clicked * 100, 2) AS click_to_completion_rate
FROM (
    SELECT
        user_id,
        session_id,
        event_type AS step
    FROM events
    WHERE event_date = today()
)
GROUP BY session_id;

Analyse de cohortes

-- Cohortes d'utilisateurs par mois d'inscription
SELECT
    toStartOfMonth(signup_date) AS cohort,
    toStartOfMonth(activity_date) AS month,
    dateDiff('month', cohort, month) AS months_since_signup,
    count(DISTINCT user_id) AS active_users
FROM (
    SELECT
        user_id,
        min(toDate(timestamp)) OVER (PARTITION BY user_id) AS signup_date,
        toDate(timestamp) AS activity_date
    FROM events
)
GROUP BY cohort, month, months_since_signup
ORDER BY cohort, months_since_signup;

Modèles de pipeline de données

Modèle ETL

// Extract, Transform, Load
async function etlPipeline() {
  // 1. Extraire de la source
  const rawData = await extractFromPostgres()

  // 2. Transformer
  const transformed = rawData.map(row => ({
    date: new Date(row.created_at).toISOString().split('T')[0],
    market_id: row.market_slug,
    volume: parseFloat(row.total_volume),
    trades: parseInt(row.trade_count)
  }))

  // 3. Charger dans ClickHouse
  await bulkInsertToClickHouse(transformed)
}

// Exécuter périodiquement
setInterval(etlPipeline, 60 * 60 * 1000)  // Toutes les heures

Capture de données modifiées (CDC)

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

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

pgClient.query('LISTEN market_updates')

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

  await clickhouse.insert({
    table: 'market_updates',
    values: [
      {
        market_id: update.id,
        event_type: update.operation,  // INSERT, UPDATE, DELETE
        timestamp: new Date(),
        data: JSON.stringify(update.new_data)
      }
    ],
    format: 'JSONEachRow'
  })
})

Bonnes pratiques

1. Stratégie de partitionnement

• Partitionner par temps (généralement par mois ou par 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é (cardinalité élevée en premier)
• L'ordre a un impact sur la compression

3. Types de données

• Utiliser le type le plus petit 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 JOINs (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

Rappelez-vous : ClickHouse excelle dans les charges de travail analytiques. Concevez les tables en fonction de vos modèles de requêtes, effectuez des insertions par lots et tirez parti des vues matérialisées pour les agrégations en temps réel.