ClickHouse 6 — Sharding & Replication

Ở bài 3 và bài 5 ta đã có bảng MergeTree, nạp dữ liệu và dựng rollup trên một node. Nhưng một cụm phân tích log/giao dịch của ngân hàng không thể sống trên một máy: dữ liệu vượt dung lượng đĩa, truy vấn cần nhiều CPU hơn một hộp cho phép, và nếu node đó chết thì cả dashboard sập. Bài này nói về mở rộng ngang (horizontal scaling) trong ClickHouse — cách trải dữ liệu và tải truy vấn ra nhiều node bằng hai trục độc lập nhưng bổ trợ nhau: replication (nhân bản) để bền vững + đọc HA, và sharding (phân mảnh) để chia dữ liệu và tính toán.

Điểm cần nắm ngay: trong ClickHouse, replication và sharding là hai cơ chế tách biệt. Replication làm việc ở tầng table engine (ReplicatedMergeTree). Sharding + định tuyến truy vấn làm việc ở tầng bảng ảo (Distributed engine). Bạn phối hợp cả hai để có một cụm vừa chịu lỗi vừa mở rộng được.

Lưu ý: mọi block SQL/XML trong bài đều là cú pháp ClickHouse, chỉ để minh hoạ. Sandbox của app là PostgreSQL read-only nên không block nào được đánh dấu "chạy được".

Hai trục mở rộng: replica vs shard

Hình dung dữ liệu như một tấm bảng lớn. Có hai cách "cắt" nó ra nhiều máy:

Một cụm production điển hình kết hợp cả hai. Ví dụ cụm 2 shard × 2 replica: dữ liệu chia làm 2 phần (shard 1, shard 2), mỗi phần được giữ 2 bản trên 2 node khác nhau. Tổng 4 node, mất bất kỳ 1 node nào cụm vẫn phục vụ đủ dữ liệu.

Replication: ReplicatedMergeTree

Mọi engine họ MergeTree đều có biến thể Replicated*: ReplicatedMergeTree, ReplicatedReplacingMergeTree, ReplicatedAggregatingMergeTree… Chúng hoạt động y hệt bản gốc về mặt lưu trữ (part, sort key, TTL — xem lại bài MergeTree) nhưng thêm khả năng tự đồng bộ dữ liệu giữa các replica.

Cơ chế: replication ở ClickHouse là theo từng bảng và bất đồng bộ (asynchronous). Khi một replica nhận INSERT và ghi xong một part mới, nó đăng ký part đó vào một replicated log đặt trong ClickHouse Keeper (hoặc ZooKeeper). Các replica khác của cùng bảng theo dõi log này, thấy part mới thì kéo (fetch) phần dữ liệu đó về qua giao thức HTTP giữa các server. Merge cũng được điều phối tương tự: một replica được chọn làm merge, kết quả nhân bản cho các replica còn lại thay vì mỗi node tự merge độc lập (tránh phí CPU trùng lặp).

Vai trò của ClickHouse Keeper (hoặc ZooKeeper) rất quan trọng và cần nói rõ: Keeper không lưu dữ liệu bảng, nó chỉ giữ metadata điều phối — danh sách replica, replicated log, thông tin part nào đã có, và các lock để bầu chọn replica làm merge/mutation. ClickHouse Keeper là bản viết lại bằng C++, tương thích giao thức ZooKeeper, gọn và nhanh hơn — hiện là lựa chọn khuyến nghị. Keeper cần chạy quorum số lẻ node (thường 3) để chịu lỗi.

Khai báo một bảng replicated cần hai tham số cho engine: đường dẫn ZooKeeper/Keeper (định danh bảng, giống nhau trên các replica cùng shard) và tên replica (khác nhau giữa các replica):

CREATE TABLE txn_local ON CLUSTER analytics_cluster
(
    txn_id      UInt64,
    account_id  UInt64,
    amount      Decimal(18, 2),
    kind        LowCardinality(String),
    created_at  DateTime
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/txn_local',   -- đường dẫn trong Keeper
    '{replica}'                               -- định danh replica
)
PARTITION BY toYYYYMM(created_at)
ORDER BY (account_id, created_at);

{shard} và {replica} là macro — mỗi node tự thay bằng giá trị của mình khai báo trong config. Nhờ vậy cùng một câu CREATE TABLE ... ON CLUSTER chạy được đồng loạt trên mọi node mà mỗi node vẫn ra đúng đường dẫn Keeper riêng. ON CLUSTER analytics_cluster bảo ClickHouse phát lệnh DDL này ra toàn bộ node trong cluster thay vì phải SSH gõ tay từng máy.

Điểm quan trọng — replication cấp bảng, không cấp instance: hai node là replica của nhau khi và chỉ khi bảng của chúng trỏ tới cùng một đường dẫn Keeper. Đây là điều làm ClickHouse khác Postgres/Oracle (nơi replication thường là cả instance — xem bài replication ở db-opt). Ở ClickHouse bạn có thể để một bảng replicated 3 bản, bảng khác chỉ 1 bản, trong cùng server.

Sharding: chia dữ liệu ra nhiều node

Replication giải quyết bền vững và đọc, nhưng mỗi replica vẫn giữ toàn bộ dữ liệu — không giúp gì khi dữ liệu vượt đĩa một máy hay khi cần nhiều CPU để quét. Đó là lúc dùng sharding: chia dữ liệu thành các tập con rời nhau, mỗi shard là một tập.

Trong ClickHouse, dữ liệu vật lý vẫn nằm trong các bảng ReplicatedMergeTree cục bộ (local table) trên từng node — mỗi shard chỉ chứa phần dữ liệu của nó. Việc "gộp" các shard lại thành một bảng logic để đọc/ghi được làm bởi một lớp ảo phía trên: Distributed engine.

Distributed table engine

Bảng Distributed là một bảng ảo, không chứa dữ liệu. Nó chỉ là bản mô tả "bảng cục bộ tên gì, nằm trên cluster nào, chia theo key gì". Mọi truy vấn tới nó được định tuyến (route) tới các bảng cục bộ trên từng shard.

CREATE TABLE txn_distributed ON CLUSTER analytics_cluster
AS txn_local
ENGINE = Distributed(
    analytics_cluster,   -- tên cluster (khai trong config)
    default,             -- database chứa bảng cục bộ
    txn_local,           -- tên bảng cục bộ trên mỗi shard
    cityHash64(account_id)  -- sharding key
);

Bốn tham số: cluster, database, bảng cục bộ, và sharding key (biểu thức tuỳ chọn). AS txn_local sao lược đồ cột từ bảng cục bộ. Từ đây, ứng dụng chỉ nói chuyện với txn_distributed như một bảng bình thường; ClickHouse lo phần còn lại.

Sharding key và phân phối INSERT

Sharding key quyết định INSERT vào txn_distributed thì mỗi dòng rơi về shard nào. ClickHouse tính sharding_key % tổng_trọng_số_shard, dựa trên weight của mỗi shard trong config, để chọn shard đích. Chọn key tốt cần:

• Phân bố đều để không bị hot shard (một shard gánh phần lớn dữ liệu). Vì thế thường bọc bằng hàm băm: cityHash64(account_id) thay vì account_id trực tiếp, tránh lệch khi ID không đều.
• Gom dữ liệu liên quan về cùng shard khi cần. Ví dụ shard theo cityHash64(account_id) khiến mọi giao dịch của một tài khoản nằm chung một shard — có lợi khi truy vấn hay JOIN theo account_id vì tránh xáo dữ liệu qua mạng.

Có hai kiểu ghi:

1. Ghi qua bảng Distributed (INSERT INTO txn_distributed): node nhận sẽ băm từng dòng, đệm tạm trong thư mục và forward bất đồng bộ tới đúng shard. Tiện nhưng thêm một chặng mạng và độ trễ đẩy; nếu node trung gian chết trước khi đẩy hết, phần đệm chưa gửi có thể mất.
2. Ghi thẳng vào bảng cục bộ trên từng shard (INSERT INTO txn_local): ứng dụng/loader tự quyết dòng nào về shard nào (thường ingest từ Kafka — xem Kafka architecture). Kiểm soát tốt hơn, thường là cách khuyến nghị cho ingest lớn.

Cluster config

Topology cụm được khai trong file cấu hình server (thường remote_servers trong config.xml hoặc file rời trong config.d/). Mỗi node trong cluster đều có bản config này. Ví dụ cụm 2 shard × 2 replica:

<remote_servers>
  <analytics_cluster>
    <shard>
      <internal_replication>true</internal_replication>
      <replica><host>node-1a</host><port>9000</port></replica>
      <replica><host>node-1b</host><port>9000</port></replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica><host>node-2a</host><port>9000</port></replica>
      <replica><host>node-2b</host><port>9000</port></replica>
    </shard>
  </analytics_cluster>
</remote_servers>

Lưu ý <internal_replication>true</internal_replication>: khi bật, INSERT qua Distributed chỉ gửi tới một replica của shard, để ReplicatedMergeTree tự nhân sang các replica còn lại. Nếu để false, Distributed sẽ tự gửi tới mọi replica — chỉ đúng khi bảng cục bộ không replicated; kết hợp sai (false + ReplicatedMergeTree) sẽ khiến dữ liệu bị ghi hai lần. Với bảng replicated, gần như luôn để true.

Macro {shard}/{replica} khai riêng cho từng node trong <macros>:

<macros>
  <shard>01</shard>
  <replica>node-1a</replica>
</macros>

Truy vấn phân tán: scatter-gather

Khi bạn SELECT từ bảng Distributed, ClickHouse chạy mô hình scatter-gather:

Node nhận truy vấn (initiator) viết lại truy vấn thành truy vấn con trên bảng cục bộ, scatter tới một replica còn sống của mỗi shard (chọn theo load_balancing), mỗi shard tự quét và tổng hợp cục bộ, trả về partial aggregate state (không phải toàn bộ dòng thô). Initiator gather, merge các state rồi finalize. Điểm hay: nhờ đẩy GROUP BY xuống shard, lượng dữ liệu qua mạng nhỏ hơn nhiều so với kéo dòng thô về gộp — đây là lý do ClickHouse quét hàng tỷ dòng trên nhiều node vẫn nhanh.

Vì scatter chỉ chọn một replica mỗi shard, việc có nhiều replica vừa cho đọc HA (replica chết thì chọn cái khác) vừa cho cân tải đọc (các truy vấn khác nhau chạm replica khác nhau).

GLOBAL IN / GLOBAL JOIN

Vấn đề nảy sinh với subquery và JOIN khi dữ liệu nằm rải rác. Xét:

SELECT count()
FROM txn_distributed
WHERE account_id IN (
    SELECT id FROM accounts_distributed WHERE city = 'Ha Noi'
);

Nếu dùng IN thường, mỗi shard khi chạy subquery SELECT id FROM accounts_distributed chỉ thấy phần accounts trên chính nó → kết quả sai, vì tài khoản Hà Nội có thể nằm ở shard khác. Giải pháp là GLOBAL IN (và tương tự GLOBAL JOIN):

SELECT count()
FROM txn_distributed
WHERE account_id GLOBAL IN (
    SELECT id FROM accounts_distributed WHERE city = 'Ha Noi'
);

Với GLOBAL IN, initiator chạy subquery một lần, thu về tập kết quả, rồi gửi tập đó kèm truy vấn con tới từng shard. Nhờ vậy mọi shard dùng chung một tập lọc đầy đủ. Đánh đổi: tập bên phải phải đủ nhỏ để truyền qua mạng và giữ trong RAM (nếu là tập khổng lồ thì cân nhắc thiết kế lại theo colocated shard key). Đây là điểm phải nhớ khi viết truy vấn phân tán — quên GLOBAL là nguồn lỗi kết quả sai âm thầm phổ biến nhất.

Đánh đổi nhất quán & insert idempotent

Replication của ClickHouse là bất đồng bộ → mô hình nhất quán là eventual consistency. Hệ quả cần biết:

• Ngay sau INSERT vào một replica, đọc từ replica khác có thể chưa thấy dữ liệu (độ trễ nhân bản, thường mili-giây tới vài giây tuỳ tải/mạng). Nếu nghiệp vụ cần "ghi xong đọc thấy ngay", có thể dùng SELECT ... SETTINGS select_sequential_consistency = 1 để buộc chọn replica đã bắt kịp, đổi lại độ trễ cao hơn và có thể lỗi nếu không replica nào kịp.
• Có thể chỉnh độ bền của INSERT bằng insert_quorum (chờ N replica xác nhận mới báo thành công) để không mất dữ liệu khi node ghi chết ngay sau đó — đổi lại độ trễ ghi.

Insert idempotent — đặc tính rất quan trọng cho ingest: ReplicatedMergeTree tự khử trùng lặp block INSERT. Mỗi block ghi vào được tính một hash; nếu cùng một block được ghi lại (ví dụ loader retry sau timeout mạng, hay Kafka consumer đọc lại offset), ClickHouse thấy hash đã có trong Keeper và bỏ qua, không tạo dữ liệu trùng. Đây là điều khiến pipeline "at-least-once" từ Kafka trở nên an toàn: cứ retry, miễn nội dung block giống hệt thì không nhân đôi. Lưu ý: điều kiện là block phải hoàn toàn giống nhau (cùng dữ liệu, cùng thứ tự) — nếu retry mà gộp/chia batch khác đi thì hash khác và sẽ ghi trùng.

So với sharding truyền thống

Ở RDBMS truyền thống (xem db-opt: partitioning & scaling), sharding thường phải làm ở tầng ứng dụng hoặc qua middleware/proxy: app tự tính dòng thuộc shard nào, tự fan-out truy vấn rồi gộp — logic phức tạp, dễ sai, và cross-shard JOIN gần như phải tự viết. ClickHouse khác ở chỗ Distributed engine gánh phần đó ở tầng database: định tuyến, scatter-gather, đẩy aggregate xuống shard đều tự động; app chỉ thấy một bảng.

Tuy vậy đừng nhầm với "distributed SQL" kiểu CockroachDB/Spanner — những hệ đó nhắm OLTP với giao dịch phân tán ACID. ClickHouse là OLAP: không giao dịch phân tán, nhất quán eventual, tối ưu cho quét/tổng hợp lượng lớn chứ không cho cập nhật từng dòng. Chọn đúng công cụ theo tải là điều cốt lõi.

ClickHouse Cloud / serverless

Mô hình sharding-replica cổ điển trên gắn chặt compute và storage vào cùng node (dữ liệu nằm trên đĩa của node tính toán). Xu hướng mới — ClickHouse Cloud và các bản serverless — tách compute khỏi storage: dữ liệu đặt trên object storage (S3/GCS) dùng chung, các node compute stateless đọc từ đó và co giãn độc lập. Khi đó bạn hầu như không phải quản lý sharding key hay số replica thủ công — hệ tự phân phối. Đánh đổi là độ trễ đọc từ object storage (bù bằng cache cục bộ) và phụ thuộc nhà cung cấp. Với cụm tự vận hành (self-managed) như phần lớn hệ thống nội bộ ngân hàng, mô hình Distributed + ReplicatedMergeTree ở trên vẫn là chuẩn.

Use case thực tế

Bối cảnh: kho log giao dịch NCB gộp core banking + internet banking + thẻ, khoảng 900 triệu dòng/ngày, giữ 18 tháng để phân tích gian lận và báo cáo. Một node không kham nổi (~500 tỷ dòng, vượt đĩa và CPU quét).

Thiết kế cụm: 4 shard × 2 replica = 8 node ClickHouse + 3 node Keeper (quorum).

1. Bảng cục bộ txn_local dùng ReplicatedMergeTree, PARTITION BY toYYYYMM(created_at), ORDER BY (account_id, created_at).
2. Sharding key cityHash64(account_id) — mọi giao dịch của một tài khoản gom về một shard, nên truy vấn "lịch sử một tài khoản" và JOIN theo account_id không xáo dữ liệu qua mạng.
3. Ingest: Kafka consumer (mỗi partition) ghi thẳng vào bảng cục bộ trên shard tương ứng, batch ~100k dòng. Consumer retry an toàn nhờ khử trùng lặp block — dù đọc lại offset cũng không nhân đôi.
4. Dashboard và báo cáo truy vấn qua bảng txn_distributed; scatter-gather đẩy GROUP BY xuống 4 shard, initiator gộp.

Kết quả (minh hoạ): truy vấn tổng hợp theo phút 24h gần nhất giảm từ ~9 giây (1 node) xuống ~2 giây (quét song song 4 shard). Mất 1 node bất kỳ, cụm vẫn phục vụ đủ dữ liệu nhờ replica. Query đối chiếu danh sách tài khoản nghi vấn dùng GLOBAL IN để tránh kết quả sai khi lọc cross-shard.

Về mặt tư duy SQL tổng hợp, đây vẫn là các câu quen thuộc:

-- Cú pháp ClickHouse (minh hoạ, chạy trên bảng Distributed)
SELECT toStartOfMinute(created_at) AS minute, kind,
       count() AS cnt, sum(amount) AS total
FROM txn_distributed
WHERE created_at >= now() - INTERVAL 24 HOUR
GROUP BY minute, kind
ORDER BY minute DESC;

Ghi nhớ

• Replication ≠ sharding. Replica sao y dữ liệu (bền vững + đọc HA); shard chia dữ liệu (dung lượng + song song). Kết hợp cả hai: N shard × M replica.
• ReplicatedMergeTree đồng bộ bất đồng bộ giữa các replica; ClickHouse Keeper/ZooKeeper giữ metadata điều phối (replicated log, lock), không giữ dữ liệu bảng. Keeper chạy quorum số lẻ (thường 3).
• Replication là cấp bảng: hai bảng là replica khi cùng trỏ một đường dẫn Keeper. Macro {shard}/{replica} giúp một câu CREATE ... ON CLUSTER dựng đúng trên mọi node.
• Distributed là bảng ảo không chứa data: định tuyến truy vấn tới bảng cục bộ trên các shard. Bốn tham số: cluster, database, bảng cục bộ, sharding key.
• Sharding key nên bọc hàm băm (cityHash64) để phân bố đều; gom dữ liệu liên quan về một shard nếu hay JOIN theo key đó.
• Với bảng replicated, để internal_replication=true để tránh ghi trùng.
• Scatter-gather: đẩy aggregate xuống shard, gộp partial state ở initiator → ít dữ liệu qua mạng.
• GLOBAL IN/GLOBAL JOIN bắt buộc khi subquery/JOIN đụng dữ liệu cross-shard, nếu không kết quả sai âm thầm.
• Nhất quán eventual: đọc replica có thể trễ; dùng select_sequential_consistency/insert_quorum khi cần chặt hơn.
• INSERT idempotent nhờ khử trùng lặp block (theo hash) — pipeline Kafka at-least-once retry an toàn, miễn block giống hệt.
• ClickHouse là OLAP, không phải distributed OLTP ACID. ClickHouse Cloud/serverless tách compute-storage, giảm việc quản sharding thủ công.