Clickhouse IO

1---
2name: clickhouse-io
3description: ClickHouse database patterns, query optimization, analytics, and data engineering best practices for high-performance analytical workloads.
4author: affaan-m
5version: "1.0"
6---
7 
8# ClickHouse Analytics Patterns
9 
10ClickHouse-specific patterns for high-performance analytics and data engineering.
11 
12## Overview
13 
14ClickHouse is a column-oriented database management system (DBMS) for online analytical processing (OLAP). It's optimized for fast analytical queries on large datasets.
15 
16**Key Features:**
17- Column-oriented storage
18- Data compression
19- Parallel query execution
20- Distributed queries
21- Real-time analytics
22 
23## Table Design Patterns
24 
25### MergeTree Engine (Most Common)
26 
27```sql
28CREATE TABLE markets_analytics (
29    date Date,
30    market_id String,
31    market_name String,
32    volume UInt64,
33    trades UInt32,
34    unique_traders UInt32,
35    avg_trade_size Float64,
36    created_at DateTime
37) ENGINE = MergeTree()
38PARTITION BY toYYYYMM(date)
39ORDER BY (date, market_id)
40SETTINGS index_granularity = 8192;
41```
42 
43### ReplacingMergeTree (Deduplication)
44 
45```sql
46-- For data that may have duplicates (e.g., from multiple sources)
47CREATE TABLE user_events (
48    event_id String,
49    user_id String,
50    event_type String,
51    timestamp DateTime,
52    properties String
53) ENGINE = ReplacingMergeTree()
54PARTITION BY toYYYYMM(timestamp)
55ORDER BY (user_id, event_id, timestamp)
56PRIMARY KEY (user_id, event_id);
57```
58 
59### AggregatingMergeTree (Pre-aggregation)
60 
61```sql
62-- For maintaining aggregated metrics
63CREATE TABLE market_stats_hourly (
64    hour DateTime,
65    market_id String,
66    total_volume AggregateFunction(sum, UInt64),
67    total_trades AggregateFunction(count, UInt32),
68    unique_users AggregateFunction(uniq, String)
69) ENGINE = AggregatingMergeTree()
70PARTITION BY toYYYYMM(hour)
71ORDER BY (hour, market_id);
72 
73-- Query aggregated data
74SELECT
75    hour,
76    market_id,
77    sumMerge(total_volume) AS volume,
78    countMerge(total_trades) AS trades,
79    uniqMerge(unique_users) AS users
80FROM market_stats_hourly
81WHERE hour >= toStartOfHour(now() - INTERVAL 24 HOUR)
82GROUP BY hour, market_id
83ORDER BY hour DESC;
84```
85 
86## Query Optimization Patterns
87 
88### Efficient Filtering
89 
90```sql
91-- ✅ GOOD: Use indexed columns first
92SELECT *
93FROM markets_analytics
94WHERE date >= '2025-01-01'
95  AND market_id = 'market-123'
96  AND volume > 1000
97ORDER BY date DESC
98LIMIT 100;
99 
100-- ❌ BAD: Filter on non-indexed columns first
101SELECT *
102FROM markets_analytics
103WHERE volume > 1000
104  AND market_name LIKE '%election%'
105  AND date >= '2025-01-01';
106```
107 
108### Aggregations
109 
110```sql
111-- ✅ GOOD: Use ClickHouse-specific aggregation functions
112SELECT
113    toStartOfDay(created_at) AS day,
114    market_id,
115    sum(volume) AS total_volume,
116    count() AS total_trades,
117    uniq(trader_id) AS unique_traders,
118    avg(trade_size) AS avg_size
119FROM trades
120WHERE created_at >= today() - INTERVAL 7 DAY
121GROUP BY day, market_id
122ORDER BY day DESC, total_volume DESC;
123 
124-- ✅ Use quantile for percentiles (more efficient than percentile)
125SELECT
126    quantile(0.50)(trade_size) AS median,
127    quantile(0.95)(trade_size) AS p95,
128    quantile(0.99)(trade_size) AS p99
129FROM trades
130WHERE created_at >= now() - INTERVAL 1 HOUR;
131```
132 
133### Window Functions
134 
135```sql
136-- Calculate running totals
137SELECT
138    date,
139    market_id,
140    volume,
141    sum(volume) OVER (
142        PARTITION BY market_id
143        ORDER BY date
144        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
145    ) AS cumulative_volume
146FROM markets_analytics
147WHERE date >= today() - INTERVAL 30 DAY
148ORDER BY market_id, date;
149```
150 
151## Data Insertion Patterns
152 
153### Bulk Insert (Recommended)
154 
155```typescript
156import { ClickHouse } from 'clickhouse'
157 
158const clickhouse = new ClickHouse({
159  url: process.env.CLICKHOUSE_URL,
160  port: 8123,
161  basicAuth: {
162    username: process.env.CLICKHOUSE_USER,
163    password: process.env.CLICKHOUSE_PASSWORD
164  }
165})
166 
167// ✅ Batch insert (efficient)
168async function bulkInsertTrades(trades: Trade[]) {
169  const values = trades.map(trade => `(
170    '${trade.id}',
171    '${trade.market_id}',
172    '${trade.user_id}',
173    ${trade.amount},
174    '${trade.timestamp.toISOString()}'
175  )`).join(',')
176 
177  await clickhouse.query(`
178    INSERT INTO trades (id, market_id, user_id, amount, timestamp)
179    VALUES ${values}
180  `).toPromise()
181}
182 
183// ❌ Individual inserts (slow)
184async function insertTrade(trade: Trade) {
185  // Don't do this in a loop!
186  await clickhouse.query(`
187    INSERT INTO trades VALUES ('${trade.id}', ...)
188  `).toPromise()
189}
190```
191 
192### Streaming Insert
193 
194```typescript
195// For continuous data ingestion
196import { createWriteStream } from 'fs'
197import { pipeline } from 'stream/promises'
198 
199async function streamInserts() {
200  const stream = clickhouse.insert('trades').stream()
201 
202  for await (const batch of dataSource) {
203    stream.write(batch)
204  }
205 
206  await stream.end()
207}
208```
209 
210## Materialized Views
211 
212### Real-time Aggregations
213 
214```sql
215-- Create materialized view for hourly stats
216CREATE MATERIALIZED VIEW market_stats_hourly_mv
217TO market_stats_hourly
218AS SELECT
219    toStartOfHour(timestamp) AS hour,
220    market_id,
221    sumState(amount) AS total_volume,
222    countState() AS total_trades,
223    uniqState(user_id) AS unique_users
224FROM trades
225GROUP BY hour, market_id;
226 
227-- Query the materialized view
228SELECT
229    hour,
230    market_id,
231    sumMerge(total_volume) AS volume,
232    countMerge(total_trades) AS trades,
233    uniqMerge(unique_users) AS users
234FROM market_stats_hourly
235WHERE hour >= now() - INTERVAL 24 HOUR
236GROUP BY hour, market_id;
237```
238 
239## Performance Monitoring
240 
241### Query Performance
242 
243```sql
244-- Check slow queries
245SELECT
246    query_id,
247    user,
248    query,
249    query_duration_ms,
250    read_rows,
251    read_bytes,
252    memory_usage
253FROM system.query_log
254WHERE type = 'QueryFinish'
255  AND query_duration_ms > 1000
256  AND event_time >= now() - INTERVAL 1 HOUR
257ORDER BY query_duration_ms DESC
258LIMIT 10;
259```
260 
261### Table Statistics
262 
263```sql
264-- Check table sizes
265SELECT
266    database,
267    table,
268    formatReadableSize(sum(bytes)) AS size,
269    sum(rows) AS rows,
270    max(modification_time) AS latest_modification
271FROM system.parts
272WHERE active
273GROUP BY database, table
274ORDER BY sum(bytes) DESC;
275```
276 
277## Common Analytics Queries
278 
279### Time Series Analysis
280 
281```sql
282-- Daily active users
283SELECT
284    toDate(timestamp) AS date,
285    uniq(user_id) AS daily_active_users
286FROM events
287WHERE timestamp >= today() - INTERVAL 30 DAY
288GROUP BY date
289ORDER BY date;
290 
291-- Retention analysis
292SELECT
293    signup_date,
294    countIf(days_since_signup = 0) AS day_0,
295    countIf(days_since_signup = 1) AS day_1,
296    countIf(days_since_signup = 7) AS day_7,
297    countIf(days_since_signup = 30) AS day_30
298FROM (
299    SELECT
300        user_id,
301        min(toDate(timestamp)) AS signup_date,
302        toDate(timestamp) AS activity_date,
303        dateDiff('day', signup_date, activity_date) AS days_since_signup
304    FROM events
305    GROUP BY user_id, activity_date
306)
307GROUP BY signup_date
308ORDER BY signup_date DESC;
309```
310 
311### Funnel Analysis
312 
313```sql
314-- Conversion funnel
315SELECT
316    countIf(step = 'viewed_market') AS viewed,
317    countIf(step = 'clicked_trade') AS clicked,
318    countIf(step = 'completed_trade') AS completed,
319    round(clicked / viewed * 100, 2) AS view_to_click_rate,
320    round(completed / clicked * 100, 2) AS click_to_completion_rate
321FROM (
322    SELECT
323        user_id,
324        session_id,
325        event_type AS step
326    FROM events
327    WHERE event_date = today()
328)
329GROUP BY session_id;
330```
331 
332### Cohort Analysis
333 
334```sql
335-- User cohorts by signup month
336SELECT
337    toStartOfMonth(signup_date) AS cohort,
338    toStartOfMonth(activity_date) AS month,
339    dateDiff('month', cohort, month) AS months_since_signup,
340    count(DISTINCT user_id) AS active_users
341FROM (
342    SELECT
343        user_id,
344        min(toDate(timestamp)) OVER (PARTITION BY user_id) AS signup_date,
345        toDate(timestamp) AS activity_date
346    FROM events
347)
348GROUP BY cohort, month, months_since_signup
349ORDER BY cohort, months_since_signup;
350```
351 
352## Data Pipeline Patterns
353 
354### ETL Pattern
355 
356```typescript
357// Extract, Transform, Load
358async function etlPipeline() {
359  // 1. Extract from source
360  const rawData = await extractFromPostgres()
361 
362  // 2. Transform
363  const transformed = rawData.map(row => ({
364    date: new Date(row.created_at).toISOString().split('T')[0],
365    market_id: row.market_slug,
366    volume: parseFloat(row.total_volume),
367    trades: parseInt(row.trade_count)
368  }))
369 
370  // 3. Load to ClickHouse
371  await bulkInsertToClickHouse(transformed)
372}
373 
374// Run periodically
375setInterval(etlPipeline, 60 * 60 * 1000)  // Every hour
376```
377 
378### Change Data Capture (CDC)
379 
380```typescript
381// Listen to PostgreSQL changes and sync to ClickHouse
382import { Client } from 'pg'
383 
384const pgClient = new Client({ connectionString: process.env.DATABASE_URL })
385 
386pgClient.query('LISTEN market_updates')
387 
388pgClient.on('notification', async (msg) => {
389  const update = JSON.parse(msg.payload)
390 
391  await clickhouse.insert('market_updates', [
392    {
393      market_id: update.id,
394      event_type: update.operation,  // INSERT, UPDATE, DELETE
395      timestamp: new Date(),
396      data: JSON.stringify(update.new_data)
397    }
398  ])
399})
400```
401 
402## Best Practices
403 
404### 1. Partitioning Strategy
405- Partition by time (usually month or day)
406- Avoid too many partitions (performance impact)
407- Use DATE type for partition key
408 
409### 2. Ordering Key
410- Put most frequently filtered columns first
411- Consider cardinality (high cardinality first)
412- Order impacts compression
413 
414### 3. Data Types
415- Use smallest appropriate type (UInt32 vs UInt64)
416- Use LowCardinality for repeated strings
417- Use Enum for categorical data
418 
419### 4. Avoid
420- SELECT * (specify columns)
421- FINAL (merge data before query instead)
422- Too many JOINs (denormalize for analytics)
423- Small frequent inserts (batch instead)
424 
425### 5. Monitoring
426- Track query performance
427- Monitor disk usage
428- Check merge operations
429- Review slow query log
430 
431**Remember**: ClickHouse excels at analytical workloads. Design tables for your query patterns, batch inserts, and leverage materialized views for real-time aggregations.