Replication, partition, sharding: scale database từ 1k lên hàng triệu user

Câu hỏi rất đúng chỗ. Mô hình hay được dạy là một master nhận write, nhiều replica nhận read. Nếu Facebook chỉ có một master cho toàn bộ dữ liệu thì chắc chắn không chịu nổi. Ý chính của bài:

Replica giải quyết READ. Muốn scale WRITE thì phải có nhiều master, mỗi master chịu trách nhiệm một phần dữ liệu: đó là sharding.

(Thuật ngữ hiện nay dùng primary/replica thay cho master/slave.) Ta đi từng bước trên shop-app.

1. Bài toán

shop-app chạy một PostgreSQL duy nhất. Lượng user tăng dần:

1k user cùng lúc     → 1 DB thoải mái
100k user            → CPU DB 80%, phần lớn là SELECT (xem sản phẩm, lịch sử đơn)
vài triệu user       → orders 500 triệu dòng, write 20k/s, một máy không đủ

2. Vấn đề

• Một máy có giới hạn CPU, RAM, IOPS. Scale up (máy to hơn) đắt dần và có trần.
• Read nhiều gấp chục lần write, nhưng cùng tranh tài nguyên với write.
• Bảng orders quá lớn: index không vừa RAM, VACUUM chạy lâu, xóa dữ liệu cũ bằng DELETE sinh hàng trăm triệu dead tuple (xem bài 6.3).
• Write vượt sức một primary.

Mỗi vấn đề có một công cụ riêng:

3. Giải pháp

3.1 Read replica (replication). Primary gửi WAL (bài 6.2) sang replica, replica replay lại. Streaming replication mặc định là asynchronous: primary commit xong là trả về, replica nhận sau vài ms tới vài giây. Khoảng trễ đó là replication lag.

Hệ quả: read-after-write. User đặt đơn (ghi primary), chuyển trang "Đơn của tôi" (đọc replica) → chưa thấy đơn vừa đặt. Cách xử lý:

• Đọc ngay sau ghi thì đọc từ primary (trong cùng request, hoặc vài giây sau khi user ghi).
• Dữ liệu cần chính xác (số dư, tồn kho khi đặt) luôn đọc primary.
• Synchronous replication cho dữ liệu không được mất, đổi lại write chậm hơn.

3.2 Partitioning: chia một bảng thành nhiều phần trong cùng một database. App vẫn query bảng orders như cũ, PostgreSQL chỉ đọc partition liên quan (partition pruning).

• Range: theo khoảng, ví dụ created_at theo tháng. Hợp nhất cho dữ liệu thời gian.
• List: theo giá trị rời rạc, ví dụ country: VN, US, JP.
• Hash: theo hash(user_id) % N, chia đều khi không có khóa tự nhiên.

3.3 Sharding: chia dữ liệu ra nhiều database server. Mỗi shard là một PostgreSQL riêng (có primary + replica riêng). Write được phân tán ra nhiều primary.

Shard key là quyết định quan trọng nhất, gần như không đổi được về sau. Với shop-app, chọn user_id: user, đơn hàng, order_items của một user nằm cùng shard, nên các query "đơn của tôi" chỉ chạm một shard.

4. Code từng bước

Bước 1: Partition bảng orders theo tháng (migration Flyway):

CREATE TABLE orders (
    id           bigint GENERATED ALWAYS AS IDENTITY,
    user_id      bigint        NOT NULL,
    status       text          NOT NULL,
    total_amount numeric(12,2) NOT NULL,
    created_at   timestamptz   NOT NULL,
    PRIMARY KEY (id, created_at)          -- PK phải chứa partition key
) PARTITION BY RANGE (created_at);

CREATE TABLE orders_2026_09 PARTITION OF orders
    FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');
CREATE TABLE orders_2026_10 PARTITION OF orders
    FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');

CREATE INDEX ON orders (user_id, created_at DESC);   -- tạo trên mọi partition

-- Xóa dữ liệu 3 năm trước: tức thì, không dead tuple
ALTER TABLE orders DETACH PARTITION orders_2023_09;
DROP TABLE orders_2023_09;

Partition mới phải được tạo trước (job định kỳ hoặc extension pg_partman).

Bước 2: Tách read/write sang replica

spring:
  datasource:
    primary:
      url: jdbc:postgresql://db-primary:5432/shop
      username: ${DB_USER}
      password: ${DB_PASSWORD}
    replica:
      url: jdbc:postgresql://db-replica:5432/shop
      username: ${DB_USER}
      password: ${DB_PASSWORD}

public class ReadWriteRoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return TransactionSynchronizationManager.isCurrentTransactionReadOnly()
                ? "replica" : "primary";
    }
}

@Configuration
public class DataSourceConfig {
    @Bean
    @ConfigurationProperties("spring.datasource.primary")
    DataSourceProperties primaryProps() { return new DataSourceProperties(); }

    @Bean
    @ConfigurationProperties("spring.datasource.replica")
    DataSourceProperties replicaProps() { return new DataSourceProperties(); }

    @Bean
    @Primary
    DataSource dataSource() {
        var routing = new ReadWriteRoutingDataSource();
        routing.setTargetDataSources(Map.of(
            "primary", primaryProps().initializeDataSourceBuilder().build(),
            "replica", replicaProps().initializeDataSourceBuilder().build()));
        routing.afterPropertiesSet();
        // Lazy: chỉ lấy connection khi chạy SQL đầu tiên, lúc đó cờ readOnly đã được set
        return new LazyConnectionDataSourceProxy(routing);
    }
}

@Transactional(readOnly = true)   // → replica
public List<OrderSummary> history(Long userId) { ... }

@Transactional                    // → primary
public Order placeOrder(PlaceOrderCommand cmd) { ... }

Bước 3: Chọn shard theo user_id (minh họa, production thường dùng thư viện/proxy như Citus, Vitess-style router):

@Component
public class ShardRouter {
    private final List<DataSource> shards;   // shard-0 .. shard-3

    public ShardRouter(List<DataSource> shards) { this.shards = shards; }

    public DataSource forUser(long userId) {
        int idx = Math.floorMod(Long.hashCode(userId), shards.size());
        return shards.get(idx);
    }
}

Bước 4: Test

-- partition pruning: chỉ đọc một partition
EXPLAIN SELECT * FROM orders
WHERE created_at >= '2026-10-01' AND created_at < '2026-11-01' AND user_id = 42;
-- Index Scan on orders_2026_10 ...   (không có orders_2026_09)

-- replication lag trên primary
SELECT client_addr, state, replay_lag FROM pg_stat_replication;

curl -X POST localhost:8080/orders -d '{"productId":1,"qty":1}' -H 'Content-Type: application/json'
curl localhost:8080/orders/me     # nếu đọc replica có thể chưa thấy đơn vừa tạo

5. Cách hoạt động

REPLICATION (scale READ)
            write
App ───────────────→ Primary ──WAL stream──→ Replica 1 ←── read
   └───────────── read ────────────────────→ Replica 2 ←── read
   write vẫn chỉ vào một Primary

PARTITIONING (một DB)
            orders  (bảng logic)
    ┌──────────┼──────────┐
 2026_08    2026_09    2026_10      ← cùng một PostgreSQL server

SHARDING (scale WRITE)
                   ShardRouter: hash(user_id) % 4
       ┌──────────────┬──────────────┬──────────────┐
   Shard 0         Shard 1        Shard 2        Shard 3
 Primary→Replica Primary→Replica Primary→Replica Primary→Replica
   user A ghi vào Shard 1 cùng lúc user B ghi vào Shard 3

Ba cách chọn shard:

HASH       shard = hash(user_id) % N        chia đều; range query theo key phải hỏi mọi shard
RANGE      user_id 1–25M → S1, 25M–50M → S2  range query dễ; dễ hotspot (user mới dồn vào shard cuối)
DIRECTORY  bảng lookup: user_id → shard     linh hoạt, di chuyển từng tenant; lookup thành điểm phụ thuộc

Consistent hashing: với hash % N, thêm shard từ 4 lên 5 làm gần như mọi key đổi shard. Consistent hashing đặt shard và key lên một vòng tròn; thêm một node chỉ di chuyển khoảng 1/N số key. Biến thể thực tế: chia sẵn nhiều slot/virtual node (ví dụ 1024 slot), map slot → shard; thêm shard chỉ cần chuyển một số slot (Redis Cluster dùng 16384 hash slot).

        0
   S3 ●    ● S0        key đi theo chiều kim đồng hồ
  ●key       ●key      tới node đầu tiên gặp
   S2 ●    ● S1
       thêm S4 giữa S0 và S1 → chỉ key nằm giữa S0..S4 di chuyển

Hotspot: một shard nóng hơn hẳn. Ví dụ shard theo seller_id và một shop lớn chiếm 30% đơn. Xử lý: chọn key có cardinality cao và phân bố đều, tách riêng tenant lớn, thêm hậu tố ngẫu nhiên cho key nóng.

Cross-shard query: "top 10 sản phẩm bán chạy toàn hệ thống" phải hỏi mọi shard rồi gộp (scatter-gather), chậm. Transaction qua nhiều shard không có ACID sẵn, phải dùng Saga/Outbox. Thường đẩy loại query này sang data warehouse hoặc read model riêng (Elasticsearch, bảng tổng hợp qua Kafka).

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Thứ tự: tối ưu query + index → cache → read replica → partition → scale up → sharding cuối cùng.
• Partition khi bảng hàng trăm triệu dòng và query/xóa theo thời gian; không partition bảng vài triệu dòng.
• Sharding chỉ khi một primary thật sự không chịu nổi write hoặc dữ liệu. Nó làm mọi thứ phức tạp vĩnh viễn.

Câu này phỏng vấn rất hay hỏi. Đừng nhảy ngay vào "sharding". Trả lời theo lộ trình: đo trước, sửa chỗ nghẽn, scale từ rẻ đến đắt.

1k user ──────────────────────────────────────────────────→ hàng triệu user

0. Đo        metric, tracing, slow query log: nghẽn ở đâu?
1. Query     index đúng, bỏ N+1, keyset pagination, bỏ SELECT *
2. App       stateless → chạy nhiều instance sau load balancer
3. Pool      HikariCP nhỏ, PgBouncer khi nhiều instance
4. Cache     Redis cho dữ liệu đọc nhiều (sản phẩm, giỏ hàng)
5. Async     Kafka/queue cho việc không cần đồng bộ (email, thống kê)
6. Replica   chuyển read sang replica, xử lý read-after-write
7. Partition orders theo tháng, archive dữ liệu cũ
8. Scale up  máy DB lớn hơn (thường rẻ hơn sharding rất nhiều)
9. Shard     theo user_id khi write vượt một primary
+ CDN cho ảnh/static, rate limiting, autoscaling, HA (multi-AZ)

Stateless vs stateful, nền tảng của bước 2:

• Stateless service: không giữ trạng thái của client trong bộ nhớ giữa các request. Mọi request tự mang đủ thông tin (JWT) hoặc trạng thái nằm ở kho chung (DB, Redis). Instance nào xử lý cũng được → thêm/bớt instance tự do.
• Stateful: giữ trạng thái cục bộ (session trong memory, file upload trên đĩa local). Request phải quay về đúng instance (sticky session), instance chết là mất trạng thái → khó scale.
• Database bản chất là stateful, nên scale DB khó hơn scale app nhiều. Đó là lý do bước 6–9 phức tạp nhất.

Ý quan trọng: một triệu request tới app không có nghĩa một triệu query xuống cùng một primary. Cache, queue, replica đã hấp thụ phần lớn.

Đúng, hashing là một phương pháp. Phân biệt bằng một câu:

Partition = chia bảng bên trong một database. Sharding = chia dữ liệu ra nhiều database/server.

PARTITION                         SHARDING
 PostgreSQL server                 Shard 1 server   Shard 2 server   Shard 3 server
   orders                            orders (A)       orders (B)       orders (C)
   ├─ P1                             users  (A)       users  (B)       users  (C)
   ├─ P2
   └─ P3
 một máy, một transaction ACID     nhiều máy, mỗi máy một phần

Sharding thường được xem là "horizontal partitioning qua nhiều node", nên hai từ hay bị dùng lẫn. Khi phỏng vấn nói rõ "partition trong một DB" hay "shard ra nhiều DB".

Hash, range, list là chiến lược quyết định một dòng thuộc phần nào. Dùng được cho cả hai:

Bạn hiểu đúng ý đầu: hash là một cách để sharding. Chỉ cần chỉnh ý sau: hash và shard không phải hai cấp độ chia lớn/nhỏ.

• Sharding = ý tưởng chia dữ liệu ra nhiều DB (WHAT).
• Shard = mỗi miếng sau khi chia, ví dụ DB-01.
• Hashing = thuật toán quyết định một record đi vào miếng nào (HOW).

100 triệu user
      │
 hash(user_id) % 4        ← hash value có rất nhiều giá trị khác nhau
      │
 ┌────┼────┬────┐
 S0   S1   S2   S3        ← nhưng cuối cùng chỉ ánh xạ vào 4 shard

User 1001 → hash → shard 2
User 1002 → hash → shard 0

Cảm giác "hash chia nhỏ" có lẽ đến từ slot/virtual node: hash ra 1024 slot nhỏ, rồi gom slot vào 4 shard. Slot là đơn vị di chuyển khi rebalance, còn shard vẫn là "miếng" vật lý. Và có thể shard mà không hash (range, geographic, directory).

Q1. Chỉ một master nhận write thì hệ thống hàng triệu user scale write thế nào?

🇻🇳 Tiếng Việt:

Read replica chủ yếu scale read, không giải quyết nút thắt write ở primary. Khi write quá lớn, em shard dữ liệu theo một shard key như user_id. Mỗi shard có primary riêng nhận write cho một phần dữ liệu và replica riêng phục vụ read, nên write được phân tán ra nhiều primary. Ngoài ra cache và message queue giảm tải và xử lý bất đồng bộ.

🇬🇧 English:

Read replicas mainly scale reads; they do not remove the write bottleneck on the primary. When write load is too high, I shard the data by a shard key such as user_id. Each shard has its own primary that handles writes for a subset of the data and its own replicas for reads, so writes are spread across many primaries. Caches and message queues also reduce load and move work to asynchronous processing.

Q2. Replication lag là gì, xử lý read-after-write thế nào?

🇻🇳 Tiếng Việt:

Replication mặc định là bất đồng bộ, replica nhận WAL chậm hơn primary một chút, đó là lag. User vừa ghi rồi đọc từ replica có thể không thấy dữ liệu của mình. Em đọc từ primary cho các thao tác ngay sau khi ghi hoặc dữ liệu cần chính xác, còn các trang chấp nhận trễ vài giây thì đọc replica. Em cũng theo dõi lag qua pg_stat_replication.

🇬🇧 English:

Replication is asynchronous by default, so replicas apply WAL slightly behind the primary; that delay is replication lag. A user who writes and then reads from a replica may not see their own change. I route reads right after a write, and reads that must be exact, to the primary, and send reads that tolerate a few seconds of staleness to replicas. I also monitor lag with pg_stat_replication.

Q3. Partition và sharding khác nhau thế nào? Chọn shard key ra sao?

🇻🇳 Tiếng Việt:

Partition chia một bảng lớn thành nhiều phần trong cùng một database, app vẫn query bảng logic và DB chỉ đọc partition liên quan. Sharding chia dữ liệu ra nhiều database server để scale write và dung lượng. Hash, range, list là chiến lược phân phối dùng được cho cả hai. Shard key tốt có cardinality cao, phân bố đều và có mặt trong hầu hết query, để mỗi request chỉ chạm một shard, ví dụ user_id cho shop.

🇬🇧 English:

Partitioning splits a large table into parts inside one database; the application still queries the logical table and the database only reads relevant partitions. Sharding spreads data across multiple database servers to scale writes and storage. Hash, range and list are distribution strategies that apply to both. A good shard key has high cardinality, distributes evenly and appears in most queries so each request hits a single shard, for example user_id in a shop.

Q4. Vì sao không dùng hash % N đơn giản? Consistent hashing giải quyết gì?

🇻🇳 Tiếng Việt:

Với hash % N, khi đổi số shard thì gần như mọi key đổi vị trí, phải di chuyển gần hết dữ liệu. Consistent hashing đặt node và key trên một vòng hash, thêm hoặc bớt một node chỉ di chuyển khoảng một phần N số key. Thực tế hay dùng số slot cố định rồi map slot vào shard, như Redis Cluster với 16384 slot.

🇬🇧 English:

With hash mod N, changing the number of shards remaps almost every key, so nearly all data must move. Consistent hashing places nodes and keys on a hash ring, so adding or removing a node only moves about one N-th of the keys. In practice systems often use a fixed number of slots mapped to shards, like Redis Cluster with 16384 slots.

Cách nhớ ngắn gọn

• Replica = scale READ; Sharding = scale WRITE.
• Partition = trong một DB; Sharding = nhiều DB.
• Sharding là WHAT, hash/range/directory là HOW.
• Async replica → lag → đọc primary sau khi ghi.
• Lộ trình: đo → query → stateless app → cache → replica → partition → scale up → shard.