Перейти до змісту

Розділ 14: Операції з базою даних

Аудиторія: DevOps, платформні інженери

Postgres у production

Cat-Scan використовує Cloud SQL (Postgres 15) як єдину операційну базу даних. API підключається через Cloud SQL Auth Proxy — сайдкар-контейнер на localhost:5432.

Ключові таблиці та масштаб

Таблиця Приблизна кількість записів Що зберігає
rtb_daily ~84 мільйони Щоденні показники RTB за покупцем, креативом, гео тощо
rtb_bidstream ~21 мільйон Деталізація потоку ставок за видавцем, гео
rtb_quality різна Метрики якості (viewability, brand safety)
rtb_bid_filtering ~188 тисяч Причини та обсяги фільтрації ставок
pretargeting_configs невелика Знімки конфігурацій претаргетингу
creatives невелика Метадані креативів та мініатюри
import_history невелика Записи імпорту CSV
users, permissions, audit_log невелика Дані автентифікації та адміністрування

Критичні індекси

Найбільш чутливий до продуктивності шаблон індексу:

CREATE INDEX idx_<table>_buyer_metric_date_desc
    ON <table> (buyer_account_id, metric_date DESC);

Він існує на таблицях rtb_daily, rtb_bidstream, rtb_quality та rtb_bid_filtering. Підтримує запит актуальності даних та аналітику в межах покупця.

Інші важливі індекси: - (metric_date, buyer_account_id): для фільтрів за діапазоном дат + покупцем - (metric_date, billing_id): для запитів у межах білінгу - (row_hash) UNIQUE: дедуплікація при імпорті

Дедуплікація

Кожен імпортований рядок хешується (стовпець row_hash). Унікальне обмеження на row_hash запобігає дублюванню вставок, що робить повторний імпорт безпечним.

Модель підключень

API використовує підключення на кожний запит (без пулу підключень). Кожен запит створює новий виклик psycopg.connect(), обгорнутий у run_in_executor для асинхронної сумісності.

async def pg_query(sql, params=()):
    loop = asyncio.get_event_loop()
    def _execute():
        with _get_connection() as conn:
            cursor = conn.execute(sql, params)
            return [dict(row) for row in cursor.fetchall()]
    return await loop.run_in_executor(None, _execute)

Для production-навантажень розгляньте додавання psycopg_pool, якщо накладні витрати на підключення стануть вузьким місцем.

Тайм-аути запитів

Для ресурсномістких запитів (наприклад, актуальність даних по великих таблицях) API використовує pg_query_with_timeout:

conn.execute(f"SET LOCAL statement_timeout = {timeout_ms}")
cursor = conn.execute(sql, params)

Ключові деталі: - SET LOCAL обмежує тайм-аут поточною транзакцією та автоматично скидається після завершення транзакції (commit або rollback). - Тайм-аут актуальності даних за замовчуванням: 30 секунд. - Налаштовується через змінну середовища UPLOADS_DATA_FRESHNESS_QUERY_TIMEOUT_MS (мінімум 1000 мс). - SET LOCAL дозволяє уникнути проблеми перерваної транзакції, яка виникає при використанні SET + RESET у блоці try/finally (якщо запит скасовано тайм-аутом, транзакція переходить у перерваний стан, і RESET завершується помилкою).

Шаблон запиту актуальності даних

Ендпоінту актуальності даних потрібно знати, за які дати є дані для кожного типу звіту. Ефективний шаблон використовує generate_series + EXISTS:

SELECT d::date AS metric_date, 'bidsinauction' AS csv_type, 1 AS row_count
FROM generate_series(%s::date, CURRENT_DATE - 1, '1 day'::interval) AS d
WHERE EXISTS (
    SELECT 1 FROM rtb_daily
    WHERE metric_date = d::date AND buyer_account_id = %s
    LIMIT 1
)

Це виконує N пошуків по індексу (один на кожний день у вікні) замість сканування мільйонів рядків. Для 14-денного вікна: 14 пошуків по ~0,1 мс кожний проти повного паралельного послідовного сканування, яке займає 160+ секунд.

Чому GROUP BY тут не працює: Навіть із 1 AS row_count (без COUNT), планувальник обирає послідовне сканування, коли набір результатів GROUP BY великий відносно таблиці. Індекс (buyer_account_id, metric_date DESC) існує, але планувальник оцінює, що сканувати 84 млн рядків дешевше, ніж виконати 4,4 млн пошуків по індексу.

Роль BigQuery

BigQuery зберігає сирі, деталізовані дані та виконує пакетні аналітичні завдання. Він не використовується для запитів API в реальному часі. Шаблон:

  1. Сирі дані CSV завантажуються в таблиці BigQuery.
  2. Пакетні завдання агрегують дані.
  3. Попередньо агреговані результати записуються в Postgres.
  4. API обслуговує запити з Postgres.

Зберігання даних

Налаштовується за адресою /settings/retention. Контролює, як довго історичні дані зберігаються в Postgres перед видаленням застарілих.

Пов'язані розділи