ClickHouse Analytics Patterns

SELECT
    toDate(timestamp)                       AS date,
    store_id,
    uniq(user_id)                           AS dau,
    sum(if(event_type = 'purchase', 1, 0))  AS daily_purchases,
    sum(if(event_type = 'purchase', revenue, 0)) AS daily_revenue,
    round(daily_purchases / dau * 100, 2)   AS purchase_rate_pct
FROM events
WHERE store_id IN ('pulso-001', 'pulso-002', 'pulso-003')
  AND timestamp >= today() - INTERVAL 30 DAY
GROUP BY date, store_id
ORDER BY date DESC, dau DESC;

-- Resultado esperado: <0.8s sobre 1.5B filas (30 días × 3 tiendas)

WITH user_cohorts AS (
    SELECT
        user_id,
        store_id,
        min(toDate(timestamp))   AS signup_date
    FROM events
    WHERE event_type = 'page_view'
      AND timestamp >= today() - INTERVAL 90 DAY
    GROUP BY user_id, store_id
),
activity AS (
    SELECT DISTINCT user_id, store_id, toDate(timestamp) AS activity_date
    FROM events
    WHERE timestamp >= today() - INTERVAL 90 DAY
)
SELECT
    toStartOfMonth(uc.signup_date)                          AS cohort_month,
    count(DISTINCT uc.user_id)                             AS cohort_size,
    countIf(dateDiff('day', uc.signup_date, a.activity_date) = 1) AS d1,
    countIf(dateDiff('day', uc.signup_date, a.activity_date) = 7) AS d7,
    countIf(dateDiff('day', uc.signup_date, a.activity_date) = 30) AS d30,
    round(d1  / cohort_size * 100, 1)                    AS ret_d1_pct,
    round(d7  / cohort_size * 100, 1)                    AS ret_d7_pct,
    round(d30 / cohort_size * 100, 1)                    AS ret_d30_pct
FROM user_cohorts uc
LEFT JOIN activity a USING (user_id, store_id)
GROUP BY cohort_month
ORDER BY cohort_month DESC;

-- 1. Tabla destino (AggregatingMergeTree)
CREATE TABLE store_stats_hourly (
    hour       DateTime,
    store_id   LowCardinality(String),
    total_events   AggregateFunction(count),
    unique_users   AggregateFunction(uniq, String),
    total_revenue  AggregateFunction(sum, Float64),
    purchases      AggregateFunction(countIf, UInt8),
    add_carts      AggregateFunction(countIf, UInt8)
) ENGINE = AggregatingMergeTree()
  PARTITION BY toYYYYMM(hour)
  ORDER BY (hour, store_id);

-- 2. Vista materializada que alimenta la tabla
CREATE MATERIALIZED VIEW store_stats_hourly_mv
TO store_stats_hourly
AS SELECT
    toStartOfHour(timestamp)              AS hour,
    store_id,
    countState()                          AS total_events,
    uniqState(user_id)                    AS unique_users,
    sumState(if(revenue > 0, revenue, 0)) AS total_revenue,
    countIfState(event_type = 'purchase') AS purchases,
    countIfState(event_type = 'add_to_cart') AS add_carts
FROM events
GROUP BY hour, store_id;

-- 3. Query del dashboard (sub-200ms para 90 días)
SELECT
    hour,
    store_id,
    countMerge(total_events)  AS events,
    uniqMerge(unique_users)   AS users,
    sumMerge(total_revenue)   AS revenue,
    countMerge(purchases)     AS purchases,
    round(countMerge(purchases) / uniqMerge(unique_users) * 100, 2) AS conv_rate
FROM store_stats_hourly
WHERE hour >= now() - INTERVAL 7 DAY
GROUP BY hour, store_id
ORDER BY hour DESC;

import { createClient } from '@clickhouse/client'

const client = createClient({
  host: process.env.CLICKHOUSE_URL,
  username: process.env.CLICKHOUSE_USER,
  password: process.env.CLICKHOUSE_PASSWORD,
  compression: { request: true, response: true },   // ClickHouse compression
  clickhouse_settings: {
    async_insert:                  1,   // buffer del servidor
    wait_for_async_insert:         0,   // fire-and-forget
    async_insert_max_data_size:   '10000000',  // 10 MB buffer
    async_insert_busy_timeout_ms: '1000'
  }
})

// Buffer circular: acumula hasta 500 eventos o cada 2 segundos
const BATCH_SIZE = 500
const FLUSH_MS   = 2000
let buffer: EventRow[] = []

export async function trackEvent(event: EventInput) {
  buffer.push({
    event_id:   crypto.randomUUID(),
    store_id:   event.storeId,
    user_id:    event.userId,
    event_type: event.type,
    session_id: event.sessionId,
    properties: event.properties ?? {},
    timestamp:  new Date().toISOString(),
    revenue:    event.revenue ?? null
  })
  if (buffer.length >= BATCH_SIZE) await flush()
}

async function flush() {
  if (!buffer.length) return
  const batch = buffer.splice(0)    // atomic swap
  await client.insert({ table: 'events', values: batch, format: 'JSONEachRow' })
}

setInterval(flush, FLUSH_MS)         // garantía temporal

SELECT
    substring(query, 1, 80)                                        AS query_preview,
    formatReadableSize(sum(read_bytes))                           AS bytes_read,
    formatReadableQuantity(sum(read_rows))                       AS rows_scanned,
    round(avg(query_duration_ms), 0)                             AS avg_ms,
    max(query_duration_ms)                                        AS max_ms,
    round(avg(memory_usage) / 1024 / 1024, 1)                  AS avg_mem_mb,
    count()                                                       AS exec_count
FROM system.query_log
WHERE type           = 'QueryFinish'
  AND query_duration_ms > 500
  AND event_time     >= now() - INTERVAL 1 HOUR
GROUP BY query_preview
ORDER BY avg_ms DESC
LIMIT 10;