Ingesting large volumes of data (batch inserts, Kafka integration)
Migrating from PostgreSQL/MySQL to ClickHouse for analytics
Implementing real-time dashboards or time-series analytics
Overview
ClickHouse is a column-oriented database management system (DBMS) for online analytical processing (OLAP). It's optimized for fast analytical queries on large datasets.
Key Features:
Column-oriented storage
Data compression
Parallel query execution
Distributed queries
Real-time analytics
Table Design Patterns
MergeTree Engine (Most Common)
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 (Deduplication)
-- For data that may have duplicates (e.g., from multiple 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 (Pre-aggregation)
-- For maintaining aggregated metrics
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
)
;
-- Query aggregated data
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
;
Query Optimization Patterns
Efficient Filtering
-- ✅ GOOD: Use indexed columns first
SELECT
*
FROM
markets_analytics
WHERE
date
>=
'2025-01-01'
AND
market_id
=
'market-123'
AND
volume
>
1000
ORDER
BY
date
DESC
LIMIT
100
;
-- ❌ BAD: Filter on non-indexed columns first
SELECT
*
FROM
markets_analytics
WHERE
volume
>
1000
AND
market_name
LIKE
'%election%'
AND
date
>=
'2025-01-01'
;
Aggregations
-- ✅ GOOD: Use ClickHouse-specific aggregation functions
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
;
-- ✅ Use quantile for percentiles (more efficient than 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
;
Window Functions
-- Calculate running totals
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
;
Data Insertion Patterns
Bulk Insert (Recommended)
import
{
ClickHouse
}
from
'clickhouse'
const
clickhouse
=
new
ClickHouse
(
{
url
:
process
.
env
.
CLICKHOUSE_URL
,
port
:
8123
,
basicAuth
:
{
username
:
process
.
env
.
CLICKHOUSE_USER
,
password
:
process
.
env
.
CLICKHOUSE_PASSWORD
}
}
)
// ✅ Batch insert (efficient)
async
function
bulkInsertTrades
(
trades
:
Trade
[
]
)
{
const
values
=
trades
.
map
(
trade
=>
`
(
'
${
trade
.
id
}
',
'
${
trade
.
market_id
}
',
'
${
trade
.
user_id
}
',
${
trade
.
amount
}
,
'
${
trade
.
timestamp
.
toISOString
(
)
}
'
)
`
)
.
join
(
','
)
await
clickhouse
.
query
(
`
INSERT INTO trades (id, market_id, user_id, amount, timestamp)
VALUES
${
values
}
`
)
.
toPromise
(
)
}
// ❌ Individual inserts (slow)
async
function
insertTrade
(
trade
:
Trade
)
{
// Don't do this in a loop!
await
clickhouse
.
query
(
`
INSERT INTO trades VALUES ('
${
trade
.
id
}
', ...)
`
)
.
toPromise
(
)
}
Streaming Insert
// For continuous data ingestion
import
{
createWriteStream
}
from
'fs'
import
{
pipeline
}
from
'stream/promises'
async
function
streamInserts
(
)
{
const
stream
=
clickhouse
.
insert
(
'trades'
)
.
stream
(
)
for
await
(
const
batch
of
dataSource
)
{
stream
.
write
(
batch
)
}
await
stream
.
end
(
)
}
Materialized Views
Real-time Aggregations
-- Create materialized view for hourly stats
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
;
-- Query the materialized view
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
;
Performance Monitoring
Query Performance
-- Check slow queries
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
;
Table Statistics
-- Check table sizes
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
;
Common Analytics Queries
Time Series Analysis
-- Daily active users
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
;
-- Retention analysis
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
;
Funnel Analysis
-- Conversion funnel
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
;
Cohort Analysis
-- User cohorts by signup month
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
;
Data Pipeline Patterns
ETL Pattern
// Extract, Transform, Load
async
function
etlPipeline
(
)
{
// 1. Extract from source
const
rawData
=
await
extractFromPostgres
(
)
// 2. Transform
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. Load to ClickHouse
await
bulkInsertToClickHouse
(
transformed
)
}
// Run periodically
setInterval
(
etlPipeline
,
60
*
60
*
1000
)
// Every hour
Change Data Capture (CDC)
// Listen to PostgreSQL changes and sync to 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
(
'market_updates'
,
[
{
market_id
:
update
.
id
,
event_type
:
update
.
operation
,
// INSERT, UPDATE, DELETE
timestamp
:
new
Date
(
)
,
data
:
JSON
.
stringify
(
update
.
new_data
)
}
]
)
}
)
Best Practices
1. Partitioning Strategy
Partition by time (usually month or day)
Avoid too many partitions (performance impact)
Use DATE type for partition key
2. Ordering Key
Put most frequently filtered columns first
Consider cardinality (high cardinality first)
Order impacts compression
3. Data Types
Use smallest appropriate type (UInt32 vs UInt64)
Use LowCardinality for repeated strings
Use Enum for categorical data
4. Avoid
SELECT * (specify columns)
FINAL (merge data before query instead)
Too many JOINs (denormalize for analytics)
Small frequent inserts (batch instead)
5. Monitoring
Track query performance
Monitor disk usage
Check merge operations
Review slow query log
Remember
ClickHouse excels at analytical workloads. Design tables for your query patterns, batch inserts, and leverage materialized views for real-time aggregations.