Настройка шардирования базы данных для веб-приложения

Настройка шардирования базы данных для веб-приложения

Шардирование нужно, когда один сервер PostgreSQL не справляется с объёмом данных или write-нагрузкой. Это не первый шаг оптимизации — до него должны быть исчерпаны индексы, партиционирование, репликация и кеширование. Но когда таблицы растут до сотен миллионов строк, а конкурентных записей тысячи в секунду — шардирование становится необходимостью.

Партиционирование vs шардирование

Партиционирование — разбивка одной таблицы на физические части внутри одного экземпляра PostgreSQL. Шардирование — распределение данных по нескольким независимым серверам.

Партиционирование проще и часто достаточно. Начинаем с него:

-- Range partitioning по дате (логи, события)
CREATE TABLE events (
    id         BIGSERIAL,
    user_id    BIGINT       NOT NULL,
    event_type VARCHAR(50)  NOT NULL,
    created_at TIMESTAMPTZ  NOT NULL,
    data       JSONB
) PARTITION BY RANGE (created_at);

CREATE TABLE events_2024_q1 PARTITION OF events
    FOR VALUES FROM ('2024-01-01') TO ('2024-04-01');

CREATE TABLE events_2024_q2 PARTITION OF events
    FOR VALUES FROM ('2024-04-01') TO ('2024-07-01');

-- Hash partitioning для равномерного распределения
CREATE TABLE user_sessions (
    id      BIGSERIAL,
    user_id BIGINT NOT NULL,
    token   VARCHAR(255) NOT NULL,
    data    JSONB
) PARTITION BY HASH (user_id);

CREATE TABLE user_sessions_0 PARTITION OF user_sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE user_sessions_1 PARTITION OF user_sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
-- и т.д. до REMAINDER 3

Шардирование через Citus

Citus — расширение PostgreSQL, превращающее его в распределённую БД. Это наименее болезненный путь к шардированию для PostgreSQL-проектов.

# Docker Compose для локального тестирования
docker run -e POSTGRES_PASSWORD=pass -p 5432:5432 citusdata/citus:12.1

-- Подключаем воркеры
SELECT citus_add_node('worker1', 5432);
SELECT citus_add_node('worker2', 5432);

-- Создаём распределённую таблицу
CREATE TABLE orders (
    id         BIGSERIAL,
    tenant_id  INT          NOT NULL,
    user_id    BIGINT       NOT NULL,
    status     VARCHAR(20)  NOT NULL,
    total      DECIMAL(12,2),
    created_at TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id, tenant_id)   -- partition key должен быть в PK
);

SELECT create_distributed_table('orders', 'tenant_id', shard_count => 32);

-- Таблица для colocation (JOIN по tenant_id будет локальным)
CREATE TABLE order_items (
    id         BIGSERIAL,
    tenant_id  INT    NOT NULL,
    order_id   BIGINT NOT NULL,
    product_id BIGINT NOT NULL,
    quantity   INT    NOT NULL,
    PRIMARY KEY (id, tenant_id)
);

SELECT create_distributed_table('order_items', 'tenant_id',
    colocate_with => 'orders');

-- Reference table: реплицируется на все воркеры
CREATE TABLE categories (id BIGSERIAL PRIMARY KEY, name VARCHAR(200));
SELECT create_reference_table('categories');

После этого запросы с фильтром по tenant_id маршрутизируются на конкретный шард. JOIN между orders и order_items по tenant_id выполняется локально на воркере.

Шардирование на уровне приложения

Когда нет Citus или нужен полный контроль — реализуем шардирование в приложении.

Shard key выбор — главное архитектурное решение. Хорошие shard keys:

• user_id — для user-centric приложений
• tenant_id — для multi-tenant SaaS
• region — для географически распределённых данных

Плохие shard keys:

• created_at — hot spot на последнем шарде
• status — неравномерное распределение
• UUID v4 — нет locality, плохой cache hit

Consistent hashing:

# sharding/router.py
import hashlib
from dataclasses import dataclass
from typing import Any

@dataclass
class ShardConfig:
    host: str
    port: int
    database: str


SHARDS: dict[int, ShardConfig] = {
    0: ShardConfig('db-shard-0', 5432, 'myapp_0'),
    1: ShardConfig('db-shard-1', 5432, 'myapp_1'),
    2: ShardConfig('db-shard-2', 5432, 'myapp_2'),
    3: ShardConfig('db-shard-3', 5432, 'myapp_3'),
}

SHARD_COUNT = len(SHARDS)


def get_shard_id(shard_key: Any) -> int:
    """Детерминированное определение шарда по ключу."""
    key_bytes = str(shard_key).encode('utf-8')
    hash_value = int(hashlib.md5(key_bytes).hexdigest(), 16)
    return hash_value % SHARD_COUNT


def get_shard_config(shard_key: Any) -> ShardConfig:
    return SHARDS[get_shard_id(shard_key)]

Подключения к шардам:

from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from functools import lru_cache


@lru_cache(maxsize=None)
def _get_engine(shard_id: int):
    cfg = SHARDS[shard_id]
    dsn = f"postgresql+psycopg2://user:pass@{cfg.host}:{cfg.port}/{cfg.database}"
    return create_engine(dsn, pool_size=5, max_overflow=10)


@contextmanager
def get_shard_session(shard_key):
    shard_id = get_shard_id(shard_key)
    Session = sessionmaker(bind=_get_engine(shard_id))
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()


# Использование:
with get_shard_session(user_id=12345) as session:
    orders = session.query(Order).filter_by(user_id=12345).all()

Cross-shard queries

Запросы через несколько шардов — самое сложное место. Два подхода:

Scatter-gather — параллельный запрос ко всем шардам, слияние на уровне приложения:

import asyncio
import asyncpg


async def get_all_orders_by_status(status: str) -> list[dict]:
    """Scatter-gather по всем шардам."""
    async def query_shard(shard_id: int) -> list[dict]:
        cfg = SHARDS[shard_id]
        conn = await asyncpg.connect(
            host=cfg.host, database=cfg.database,
            user='app', password='pass'
        )
        rows = await conn.fetch(
            "SELECT * FROM orders WHERE status = $1 ORDER BY created_at DESC LIMIT 100",
            status
        )
        await conn.close()
        return [dict(r) for r in rows]

    results = await asyncio.gather(*[
        query_shard(i) for i in range(SHARD_COUNT)
    ])

    # Merge + sort
    all_orders = [o for shard_result in results for o in shard_result]
    all_orders.sort(key=lambda x: x['created_at'], reverse=True)
    return all_orders[:100]

Global index — отдельная таблица маппинга в выделенной БД:

-- В отдельной "routing" базе данных
CREATE TABLE order_shard_map (
    order_id  BIGINT  PRIMARY KEY,
    shard_id  INT     NOT NULL,
    user_id   BIGINT  NOT NULL
);
CREATE INDEX ON order_shard_map (user_id);

При создании заказа — записываем маппинг. При поиске по order_id — сначала находим шард, потом запрашиваем конкретный.

Решардирование

Добавление нового шарда — болезненная операция без Citus. Consistent hashing с виртуальными узлами (vnodes) минимизирует перемещение данных:

Вместо hash(key) % N
Используем: найти ближайший vnode на кольце из 150 виртуальных узлов
При добавлении шарда: перемещается ~1/N данных, не 1-(1/N)

Citus решает это автоматически через citus_rebalance_start().

Сроки

Настройка партиционирования PostgreSQL для существующей таблицы: 1–2 дня. Установка и настройка Citus для нового проекта: 2–3 дня. Реализация application-level шардирования со scatter-gather и global index: 3–5 дней.