Kafka Partitioning ↔ ClickHouse Sort-Key Alignment — a hands-on lab

---

A reproducible, end-to-end lab that proves how a Kafka partitioning strategy turns into the part-level pruning behavior of a ClickHouse MergeTree — and how the ingestion path (Kafka table engine vs Kafka Connect Sink) changes whether that locality survives.

Same data. Same ORDER BY (customer_id, ts). Same index_granularity. The only variables are (1) how records map to Kafka partitions (range vs hash) and (2) which ingestion path carries them into ClickHouse. Everything is measured with EXPLAIN ESTIMATE, EXPLAIN indexes = 1, and system.query_log.

🎯 What this lab proves

ClickHouse always sorts rows inside a part by the table's ORDER BY, so granule skipping within a part works regardless. The difference shows up at the part level: each part stores the min/max of the sort key, and the planner skips whole parts that can't contain the value. Whether parts are skippable depends on whether each part holds a narrow, disjoint key range — and that is decided upstream, by how Kafka partitions are formed and how the ingestion path batches them into parts.

Three ingestion paths × two partitioning strategies:

Path	How it batches partitions → parts	Locality result
A. Kafka engine, DEFAULT	one consumer multiplexes all partitions into each INSERT block	diluted — even a range topic yields full-range parts
B. Kafka engine, TUNED	kafka_num_consumers = N + kafka_thread_per_consumer = 1 → one consumer per partition, parallel flush	preserved — part = one partition's range
C. Kafka Connect Sink	groups records by topic-partition, one batch per partition (by design)	preserved with no CH tuning, but many tiny parts → rely on merges

ClickPipes is intentionally out of scope here (managed, batches by size/time; the same MergeTree benefit applies if the batch reflects partition locality, but that isn't a documented guarantee — verify with per-part min/max if you use it).

🧪 The experiment

• 10M events per topic, customer_id ∈ [0, 9999], 8 partitions.
• events_misaligned — producer uses the default (hash) partitioner.
• events_aligned — producer assigns partition = customer_id // 1250 (range).
• Both topics get the identical interleaved time order; only the partitioner differs. (This "all customers active at once" timeline is exactly what exposes the engine-default dilution.)
• Background merges are stopped so the streaming steady-state (many unmerged parts) stays observable — the regime where part-level pruning actually matters.

📊 Verified results (ClickHouse 26.6.1, full raw output in RESULTS.md)

Per-part customer_id span — lower = better pruning:

target	path · topic	parts	avg key span	reads for id=4242
engine_aligned	B engine tuned · range	38	1249	5 parts / 40,960 rows
engine_misaligned	B engine tuned · hash	32	9992	32 parts / 262,144 rows
engine_default_aligned	A engine default · range	10	9999 ⚠	10 parts / 82,418 rows
connect_aligned	C Connect · range	1861	1249	234 parts / 996,619 rows †
connect_misaligned	C Connect · hash	1836	9989	1836 parts / 7,839,808 rows

EXPLAIN indexes = 1 (granule-level): engine_aligned reads 5/38 parts, 5/1216 granules; engine_misaligned reads 32/32 parts, 32/1220 granules.

Three lessons: 1. Range partitioning works — engine_aligned keeps each part to one 1,250-wide bucket, so a point lookup skips 33 of 38 parts. 2. The engine default is a trap — engine_default_aligned consumes the same range topic but a single consumer multiplexes partitions into each block, so every part spans the full range (span 9999). You must set kafka_thread_per_consumer = 1 + kafka_num_consumers = partitions. 3. Connect keeps locality for free, but explodes part count — connect_aligned has narrow parts (span 1249) with zero CH tuning, but 1,861 tiny parts. †With merges off the lookup still touches 234 of them. Enable merges → OPTIMIZE … FINAL collapses it to 1 part / 8,192 rows; merges preserve ORDER BY, so locality survives. Connect's lever is healthy merges, not consumer tuning.

📁 File structure

kafka-partitioning-ingestion/
├── README.md                  # this file
├── RESULTS.md                 # full raw output of the verified run
├── docker-compose.yml         # Kafka (KRaft) + ClickHouse + Kafka Connect
├── connect.Dockerfile         # Connect worker + ClickHouse sink connector
├── .env                       # ports + experiment dimensions (ROWS, partitions…)
├── requirements.txt           # confluent-kafka (producer)
├── _env.sh                    # shared .env loader + `chc` helper
├── 00-explain-estimate.sql    # deterministic SQL-only proof (no Kafka)
├── 01-up.sh                   # build + start the stack, wait for healthy
├── 02-create-topics.sh        # two 8-partition topics
├── 03-clickhouse-setup.sql    # 5 target tables + engine consumers (A/B) + MVs + user
├── 04-register-connectors.sh  # register the two Connect sinks (C)
├── 05-produce.py              # produce 10M events/topic, range vs hash partitioner
├── 06-compare.sql             # the full comparison (rows, span, ESTIMATE, query_log)
├── 07-benchmark.sh            # wall-clock the point lookup across all targets
└── 99-cleanup.sh              # tear down

🏗️ Architecture

                              ┌────────────────────── ClickHouse ──────────────────────┐
events_aligned (range) ──┬──► Kafka engine (tuned, 8 consumers) ─MV─► engine_aligned
   8 partitions          ├──► Kafka engine (DEFAULT, 1 consumer) ─MV─► engine_default_aligned
                         └──► Connect Sink (8 tasks) ───────────────► connect_aligned
events_misaligned(hash)──┬──► Kafka engine (tuned, 8 consumers) ─MV─► engine_misaligned
   8 partitions          └──► Connect Sink (8 tasks) ───────────────► connect_misaligned
                              └─ all targets: MergeTree ORDER BY (customer_id, ts) ─────┘

🚀 Run it

Prereqs: Docker (with Compose) and Python 3.9+. First 01-up.sh builds the Connect image (installs the connector) — a few minutes; later runs are instant.

cd workload/kafka-partitioning-ingestion

./01-up.sh                                                            # start Kafka + ClickHouse + Connect
./02-create-topics.sh                                                 # 2 topics × 8 partitions
docker compose exec -T clickhouse clickhouse-client --multiquery < 03-clickhouse-setup.sql
./04-register-connectors.sh                                           # the two Connect sinks

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python 05-produce.py                                                  # 10M rows → each topic (~1 min)

# wait a few seconds for all consumers to drain, then:
docker compose exec -T clickhouse clickhouse-client --multiquery < 06-compare.sql
./07-benchmark.sh

# deterministic SQL-only proof (no Kafka needed):
docker compose exec -T clickhouse clickhouse-client --multiquery < 00-explain-estimate.sql

./99-cleanup.sh            # or: ./99-cleanup.sh --hard   (also wipes volumes)

Tune the experiment in .env (ROWS, NUM_CUSTOMERS, NUM_PARTITIONS, LOOKUP_CUSTOMER). If you change NUM_PARTITIONS, also update the hard-coded kafka_num_consumers in 03-clickhouse-setup.sql. Ports default to 8124 (ClickHouse HTTP), 9001 (native), 9094 (Kafka), 8083 (Connect) to avoid colliding with the langfuse lab.

⚠️ When NOT to over-engineer this

Range/custom Kafka partitioning buys part-level pruning for point/equality lookups on the partition key, but it brings real operational cost: hot partitions, skew, and a partition count that is hard to change later. If your queries are mostly time-range scans, prefer ORDER BY (time, …) with ordinary hash partitioning — that already prunes by time, and you avoid the skew risk. And remember: once parts are merged, within-part granule skipping handles point lookups anyway. The alignment win is specifically about the unmerged, high-throughput streaming steady-state.

🧠 Key takeaways

• A Kafka partition strategy is implicitly a ClickHouse part-layout strategy.
• Plain hash(key) keeps per-key order but mixes keys into every partition, so parts span the full range and can't be pruned. Range/custom partitioning is what produces narrow, skippable parts.
• The ingestion path matters as much as the partitioner: the Kafka engine dilutes locality unless you set kafka_thread_per_consumer = 1 + kafka_num_consumers = partitions; Connect Sink preserves it by design but needs healthy merges to tame part count.
• Always verify with EXPLAIN indexes = 1 (or per-part min/max) — design intent is not proof.

📝 License

MIT — same as the rest of the repository.

👤 Author

Ken Lee (ClickHouse Solution Architect) — ken.lee@clickhouse.com 작성일: 2026-06-28

---

kafka-partitioning-ingestion/
├── README.md                  # 이 문서
├── RESULTS.md                 # 검증 실행의 전체 raw 출력
├── docker-compose.yml         # Kafka (KRaft) + ClickHouse + Kafka Connect
├── connect.Dockerfile         # Connect 워커 + ClickHouse 싱크 커넥터
├── .env                       # 포트 + 실험 차원 (ROWS, partitions…)
├── requirements.txt           # confluent-kafka (프로듀서)
├── _env.sh                    # .env 로더 + `chc` 헬퍼
├── 00-explain-estimate.sql    # 결정적 SQL-only 증명 (Kafka 불필요)
├── 01-up.sh                   # 스택 빌드+기동, healthy 대기
├── 02-create-topics.sh        # 8 파티션 토픽 2개
├── 03-clickhouse-setup.sql    # 타겟 5개 + 엔진 컨슈머(A/B) + MV + 유저
├── 04-register-connectors.sh  # Connect 싱크 2개 등록 (C)
├── 05-produce.py              # 토픽당 1천만 이벤트, range vs hash 파티셔너
├── 06-compare.sql             # 전체 비교 (행, 폭, ESTIMATE, query_log)
├── 07-benchmark.sh            # 전 타겟 point lookup 시간 측정
└── 99-cleanup.sh              # 정리

cd workload/kafka-partitioning-ingestion

./01-up.sh                                                            # Kafka + ClickHouse + Connect 기동
./02-create-topics.sh                                                 # 토픽 2 × 파티션 8
docker compose exec -T clickhouse clickhouse-client --multiquery < 03-clickhouse-setup.sql
./04-register-connectors.sh                                           # Connect 싱크 2개

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python 05-produce.py                                                  # 각 토픽에 1천만 행 (~1분)

# 모든 컨슈머가 소비를 끝낼 때까지 수 초 대기 후:
docker compose exec -T clickhouse clickhouse-client --multiquery < 06-compare.sql
./07-benchmark.sh

# 결정적 SQL-only 증명 (Kafka 불필요):
docker compose exec -T clickhouse clickhouse-client --multiquery < 00-explain-estimate.sql

./99-cleanup.sh            # 또는: ./99-cleanup.sh --hard   (볼륨까지 삭제)