פיצול אופקי עבור יישומי PostgreSQL בעלי עומס גבוה

שימו לב: כאשר מספר הרשומות בטבלת ההזמנות עולה על 200 מיליון ועומס הכתיבה מגיע ל-5000 עסקאות בשנייה, PostgreSQL על שרת יחיד לא יכול להתמודד עם זה: זמן האחזור גדל ל-100 אלפיות השנייה, נקודות הביקורת מאטות לכמה דקות, הדיסק מלא (10 TB). כבר ניסיתם חלוקה לפי תאריך, שכפול מאסטר-סלייב וקאש Redis — אבל קונפליקטים בכתיבה ונעילות נשארים. האפשרות היחידה שנותרה היא חלוקה אופקית של מסד הנתונים (sharding). אנו מתכננים ומיישמים פתרונות כאלה עבור יישומי אינטרנט בעלי עומס גבוה. הניסיון שלנו: מעל 50 פרויקטים עם מערכות מבוזרות, 8 שנות ניסיון, ואנו מבטיחים אמינות. הזמינו ביקורת מסד נתונים ב-$499 — נמצא את צווארי הבקבוק ונציע את הארכיטקטורה האופטימלית, ובכך נחסוך לכם עד 40% בעלויות התשתית. חיסכון טיפוסי: $5,000 לחודש בתשתית.

חלוקה (Partitioning) מול Sharding: מה לבחור?

חלוקה מפצלת טבלה בודדת לחלקים פיזיים בתוך מופע PostgreSQL אחד. Sharding מפיץ נתונים על פני מספר שרתים עצמאיים. חלוקה פשוטה פי 10 מ-Sharding ולעיתים קרובות מספיקה — התחילו איתה. לפי תיעוד PostgreSQL, חלוקה מומלצת לטבלאות מעל 100 GB.

-- 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);
-- и т.д. до REMAINDER 3

אם חלוקה כבר לא עוזרת (עומס הכתיבה פוגע במעבד, הנתונים לא נכנסים לדיסק), עברו ל-Sharding.

איך לבחור את מפתח ה-Shard?

מפתח ה-Shard הוא ההחלטה הארכיטקטונית המרכזית. אפשרויות טובות: -- 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); -- и т.д. до REMAINDER 3 עבור יישומים ממוקדי משתמש, user_id עבור SaaS רב-דיירים, tenant_id עבור נתונים מבוזרים גיאוגרפית. אפשרויות רעות: region — נקודה חמה ב-shard האחרון, created_at — חלוקה לא אחידה, UUID v4 — ללא לוקאליות, פגיעה בקאש.

למה להשתמש ב-Citus במקום Sharding מותאם אישית?

Citus הוא הרחבה של PostgreSQL שהופכת אותו למסד נתונים מבוזר. Citus מקצר את זמן היישום פי 5 בהשוואה ל-Sharding מותאם אישית, מכיוון שהוא מנהל אוטומטית את החלוקה, האיזון מחדש והלוקאליות של JOIN. רישיון Citus Enterprise עולה כ-$1,000 לחודש, אבל החיסכון בתשתית יכול להגיע ל-$5,000 לחודש בשל הפחתת מספר השרתים ב-30%.

-- Подключаем воркеры
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)
);
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');

לאחר מכן, שאילתות עם פילטר על status מנותבות ל-shard הספציפי. JOIN בין -- Подключаем воркеры 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) ); 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 לפי orders מתבצעים מקומית על העובד (worker).

השוואת גישות:

Sharding מותאם אישית: מתי לשליטה מלאה?

ללא Citus (או כשצריך שליטה מלאה), אנו מיישמים Sharding ברמת היישום. השתמשו בגיבוב עקבי (consistent hashing) עם 150 צמתים וירטואליים — זה ממזער את תנועת הנתונים במהלך resharding ל-1/N בלבד מהנתונים.

# 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)]

חיבורים ל-shards:

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()

איך להתמודד עם שאילתות ללא מפתח Shard?

שאילתות ללא מפתח Shard הן הקשות ביותר. קיימות שתי גישות. Scatter-gather שולח שאילתות לכל ה-shards במקביל: יישום פשוט, אבל זמן האחזור גדל ליניארית עם כל shard חדש. אינדקס גלובלי שומר את המיפוי במסד נתונים נפרד: חיפוש מהיר, אבל תקורה כפולה בכתיבה. Scatter-gather מתאים לשאילתות אנליטיות נדירות (למשל, פעם בשעה); אינדקס גלובלי עדיף אם שאילתות חוצות-shards מתרחשות ביותר מ-10% מהמקרים.

import asyncio
import asyncpg

async def get_all_orders_by_status(status: str) -> list[dict]:
    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)])
    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]

תהליך העבודה

1. ניתוח העומס הנוכחי וצווארי הבקבוק: מדידת תפוקת כתיבה, זמן אחזור, גודל מסד הנתונים, דפוסי שאילתות.
2. עיצוב הסכימה: בחירת מפתח Shard, מספר ה-shards, אסטרטגיית שכפול.
3. פיתוח הנתב והעברת נתונים: יישום ניתוב (Citus או ברמת היישום), העברת נתונים עם זמן השבתה מינימלי.
4. בדיקות עומס: הדמיית עומס שיא, אימות זמן אחזור ותפוקה.
5. פריסה וניטור: הגדרת התראות לנקודות חמות, שאילתות איטיות, כשלי איזון מחדש.

Resharding: איך להוסיף Shard חדש ללא זמן השבתה?

עם גיבוב עקבי וצמתים וירטואליים (vnodes), רק ~1/N מהנתונים זזים (למשל, 25% בעת הוספת shard רביעי ל-3). Citus מפיץ מחדש נתונים אוטומטית באמצעות order_items. ללא Citus, התהליך מורכב יותר: עצירת היישום (אפשרו 30 דקות זמן השבתה), הפצת נתונים מחדש על פני הטבעת החדשה, עדכון תצורת הנתב. כדי למזער זמן השבתה, השתמשו בהגירה הדרגתית עם shards ישנים לקריאה בלבד (זמן השבתה < 5 דקות).

מה כלול

• תרשים ארכיטקטוני של מסד הנתונים המבוזר עם מפתחות Shard ותוכנית ניתוב.
• תצורת Shard (הגדרות PostgreSQL, מאגרי חיבורים, ניטור).
• יישום נתב ברמת היישום או באמצעות Citus.
• הגדרת ניטור (Prometheus + Grafana) לנקודות חמות וזמן אחזור.
• תיעוד תפעול ושחזור.
• הדרכת צוות על סכימה מבוזרת.
• תמיכה ל-30 יום לאחר ההשקה.

coordinator: 4 vCPU, 16 GB RAM, SSD
worker1: 8 vCPU, 32 GB RAM, NVMe
worker2: 8 vCPU, 32 GB RAM, NVMe
shard_count: 64
replication_factor: 2

לוחות זמנים ומחירים משוערים

המחירים מחושבים באופן אישי. חיסכון בתשתית מ-Sharding נכון יכול להגיע ל-40% — למשל, הפחתת חשבון AWS מ-$10k ל-$6k לחודש. קבלו ייעוץ (30 דקות) — ננתח את העומס שלכם ונציע את הארכיטקטורה האופטימלית. אנו מספקים פתרונות Sharding סוהר תוך 3-5 ימים. צרו קשר להערכת הפרויקט.

אנו ממליצים גם לקרוא על גיבוב עקבי ותיעוד Citus.