Інженерний підхід до зберігання спарсених даних: PostgreSQL, версіонування, JSONB

Інженерний підхід до зберігання спарсених даних: PostgreSQL, версіонування, JSONB

Ви зібрали 10 000 оголошень з Avito, а за тиждень половина змінилася. CSV тут же перетвориться на кашу — зміни не відстежити. MongoDB без строгої схеми — лише відтермінування хаосу: з часом дані забруднюються. Потрібна система, яка запам'ятає кожну зміну і дозволить знайти потрібне за мілісекунди. Ми створюємо такі рішення — на базі PostgreSQL з версіонуванням, повнотекстовим пошуком та REST API. Наша схема десятиліттями працює в продакшені. Документація PostgreSQL рекомендує GIN-індекси для JSONB, що дає швидкість і гнучкість. Давайте розберемо конкретну реалізацію. Основна таблиця scraped_items зберігає актуальні дані, а scraped_items_history — архів змін. Такий підхід забезпечує повний audit trail без втрати продуктивності.

Схема зберігання в PostgreSQL

-- Основна таблиця з історією змін CREATE TABLE scraped_items ( id BIGSERIAL PRIMARY KEY, source_id INTEGER REFERENCES sources(id), external_id TEXT NOT NULL, -- ID на стороні джерела url TEXT NOT NULL, data JSONB NOT NULL, -- гнучка схема для різних джерел data_hash CHAR(64) NOT NULL, -- SHA-256 від data для детекції змін first_seen TIMESTAMPTZ DEFAULT NOW(), last_seen TIMESTAMPTZ DEFAULT NOW(), changed_at TIMESTAMPTZ, UNIQUE (source_id, external_id) ); -- Історія змін CREATE TABLE scraped_items_history ( id BIGSERIAL PRIMARY KEY, item_id BIGINT REFERENCES scraped_items(id), data JSONB NOT NULL, recorded_at TIMESTAMPTZ DEFAULT NOW() ); -- Індекси CREATE INDEX ON scraped_items USING GIN (data); -- пошук по JSONB CREATE INDEX ON scraped_items (source_id, last_seen); CREATE INDEX ON scraped_items USING GIN ( to_tsvector('russian', data->>'title' || ' ' || COALESCE(data->>'description', '')) ); 

Ключові рішення: використання JSONB для змінної схеми, окрема таблиця історії, хеш даних для швидкого виявлення змін. Така схема забезпечує ACID-транзакції та цілісність при паралельному завантаженні.

Логіка оновлення

def upsert_item(source_id, external_id, url, data): data_hash = hashlib.sha256( json.dumps(data, sort_keys=True).encode() ).hexdigest() existing = db.query( 'SELECT id, data_hash FROM scraped_items WHERE source_id=%s AND external_id=%s', (source_id, external_id) ).fetchone() if existing is None: # новий елемент db.execute( 'INSERT INTO scraped_items (source_id, external_id, url, data, data_hash) ' 'VALUES (%s, %s, %s, %s, %s)', (source_id, external_id, url, json.dumps(data), data_hash) ) elif existing['data_hash'] != data_hash: # дані змінилися — зберігаємо історію db.execute( 'INSERT INTO scraped_items_history (item_id, data) ' 'SELECT id, data FROM scraped_items WHERE id=%s', (existing['id'],) ) db.execute( 'UPDATE scraped_items SET data=%s, data_hash=%s, last_seen=NOW(), changed_at=NOW() ' 'WHERE id=%s', (json.dumps(data), data_hash, existing['id']) ) else: # дані не змінилися — оновлюємо only last_seen db.execute( 'UPDATE scraped_items SET last_seen=NOW() WHERE id=%s', (existing['id'],) ) 

Функція upsert_item обробляє три сценарії: вставка нового об'єкта, оновлення зі збереженням історії (якщо хеш змінився), і лише позначка last_seen без змін. Це мінімізує I/O і прискорює обробку. Під навантаженням 100 000 записів на день весь pipeline вкладається в 15 хвилин.

Чому JSONB, а не окремі колонки?

Спарсені дані часто мають нестабільну структуру: сьогодні у товару є вага, завтра — колір. JSONB знімає проблему міграцій і дозволяє індексувати будь-які поля через GIN-індекс. Ми використовуємо гібридний підхід: ключові поля виносимо в колонки для швидких фільтрів, а решту — в JSONB. Це дає швидкість реляційної моделі та гнучкість документо-орієнтованої. На практиці JSONB в PostgreSQL працює в 3 рази швидше за MongoDB при вибірках за структурованими полями.

Як ми детектуємо зміни без втрати продуктивності?

Використовуємо SHA-256 від серіалізованого JSON. Хеш порівнюється зі збереженим при кожному upsert. Така перевірка виконується за O(1) і не потребує читання всього рядка. Для великих обсягів (мільйони записів) застосовуємо партиціонування за source_id.

Порівняння підходів до зберігання

Pipeline обробки: етапи та інструменти

Процес реалізації

1. Аналіз джерел — визначаємо структуру даних та частоту оновлення.
2. Проектування схеми — обираємо індекси, налаштовуємо партиціонування для великих обсягів.
3. Розробка upsert-логіки — пишемо функцію з детекцією змін за хешем.
4. Pipeline обробки — нормалізація, збагачення, агрегація.
5. API та експорт — REST endpoints з пагінацією, а також вивантаження в CSV/XLSX.
6. Моніторинг та архівування — TTL-політика, сповіщення про збої.

Що входить в роботу

• Спроектована схема БД з міграціями
• GitHub-репозиторій з кодом (upsert, pipeline, API)
• Документація по API (OpenAPI/Swagger)
• Інструкція з розгортання (Docker Compose)
• Гарантія відсутності помилок на етапі приймання

Архівація та TTL

Старі дані (не бачені більше 90 днів) переводяться в архів або видаляються — залежить від вимог. Історія змін зберігається довше за основні дані — за замовчуванням 365 днів. Все налаштовується під ваш бізнес-кейс.

Експорт

• CSV/XLSX — через pandas.to_excel() або csv.DictWriter
• REST API — FastAPI/Laravel з фільтрацією, пагінацією, сортуванням
• Webhook — відправка нових/змінених записів в сторонню систему в реальному часі

Час реалізації системи зберігання з історією змін та API — 4–6 днів. Ми гарантуємо коректну роботу під навантаженням до 100 тис. записів на день. Якщо вам потрібне надійне сховище спарсених даних — зв'яжіться з нами для обговорення схеми. Наш досвід — десятки впроваджених систем. Отримайте консультацію — ми допоможемо підібрати оптимальну схему.