Інкрементальний імпорт товарів: лише зміни

Уявіть: ви керуєте інтернет-магазином з каталогом 300 000 товарів. Щоночі запускається повний імпорт — сервери завантажені на 100%, база даних блокується, а клієнти скаржаться на неактуальні ціни та затримки в обробці замовлень. При цьому реально змінюється не більше 5% асортименту — решта 95% завантажуються марно. Така ситуація знайома багатьом e-commerce проектам.

Ми вирішуємо цю проблему за допомогою інкрементального імпорту товарів: обробляємо лише змінені позиції. За кілька років ми впровадили таку схему для 15 проектів — від невеликих магазинів до маркетплейсів з каталогом до 500 000 SKU. Наш досвід гарантує, що ви отримаєте працююче рішення в стислі терміни. Вартість впровадження починається від $3000 і окупається за 2–3 місяці за рахунок економії на серверах (до $1000 на місяць).

Інкрементальний імпорт (delta import) — це метод синхронізації, при якому оновлюються лише змінені записи. Це знижує навантаження на сервер на 80% і скорочує час імпорту в 10 разів порівняно з повним перезавантаженням. Наприклад, на одному з проектів з каталогом 200 000 SKU ми скоротили час синхронізації з 3 годин до 18 хвилин, використовуючи timestamp та хеш-порівняння для надійності. Інкрементальний імпорт товарів — це рішення, яке економить до 90% часу.

Як вибрати стратегію визначення змін?

Вибір методу залежить від можливостей постачальника даних. Порівняємо основні підходи в таблиці:

Розглянемо кожен метод детальніше.

Часова мітка

Найпоширеніший підхід — постачальник підтримує фільтр за датою зміни: GET /api/products?updated_after=2024-01-15T10:00:00Z. Система запам'ятовує час останньої успішної синхронізації і передає його при наступному запиті. Цей метод простий, але може пропустити зміни, якщо оновлення відбулося під час обробки. Тому ми завжди фіксуємо час початку синхронізації, а не кінця.

Курсор / change log

Постачальник веде лог змін зі зростаючим ID. Більш надійно, ніж timestamp: не пропускає зміни, що відбулися під час обробки. Приклад: GET /api/changes?since_id=48291. Підходить для API з високою частотою оновлень. Курсорний метод у 1.5 рази точніший за timestamp.

Хеш-порівняння

Зауважимо: коли джерело не підтримує фільтрацію за змінами — порівнюємо хеш рядка:

$hash = md5(serialize([ $row['price'], $row['qty'], $row['name'], $row['description'] ])); 

Рядок обробляється лише якщо хеш змінився. Цей метод хороший, коли дані надходять повним вивантаженням, але ми хочемо обробити тільки змінені записи.

Diff-файли

Постачальник публікує щогодинний diff-файл замість повного прайсу:

<changes> <updated id="SKU-123"><price>4990</price><qty>15</qty></updated> <updated id="SKU-456"><qty>0</qty></updated> <deleted id="SKU-789"/> <created id="SKU-999"><!-- повні дані --></created> </changes> 

Цей метод найбільш точний, але вимагає підтримки з боку постачальника.

Як ми реалізуємо інкрементальний імпорт товарів?

State Tracker

Стан синхронізації зберігається в БД. Ми використовуємо таблицю import_sync_state:

CREATE TABLE import_sync_state ( source_id int PRIMARY KEY REFERENCES import_sources(id), last_sync_at timestamptz, last_cursor varchar(200), last_change_id bigint, items_synced bigint DEFAULT 0, updated_at timestamptz DEFAULT now() ); 

Фіксуємо час початку синхронізації, а не кінця. Якщо за час обробки з'явилися нові зміни — вони потраплять у наступний цикл.

Pipeline інкрементального імпорту

Ключовий клас, який виконує синхронізацію:

class IncrementalImportJob implements ShouldQueue { public function handle( SyncStateManager $state, SupplierApiClient $client, IncrementalProductSync $sync, ): void { $since = $state->getLastSyncAt($this->sourceId); $state->markSyncStarted($this->sourceId); $stats = ['created' => 0, 'updated' => 0, 'deleted' => 0, 'skipped' => 0]; foreach ($client->fetchUpdatedSince($since) as $item) { $result = $sync->process($item, $this->sourceId); $stats[$result]++; } $state->markSyncCompleted($this->sourceId); $this->logResult($stats); } } 

Виявлення видалених позицій

Якщо джерело не надсилає явних сигналів про видалення — використовуємо anti-join через тимчасову таблицю (для каталогів від 50 000 SKU):

CREATE TEMP TABLE current_import_skus (sku varchar(100)); COPY current_import_skus FROM STDIN; UPDATE products SET deleted_at = now() WHERE source_id = $1 AND deleted_at IS NULL AND sku NOT IN (SELECT sku FROM current_import_skus); DROP TABLE current_import_skus; 

Захист від подвійного запуску

Використовуємо розподілене блокування через кеш (Redis). Якщо синхронізація вже виконується для даного джерела — новий запуск пропускається. Час життя блокування — 1 година, цього достатньо для більшості каталогів. Це запобігає дублюванню та конфліктам.

Кроки впровадження

1. Аналіз API постачальника та вибір методу визначення змін.
2. Розробка state tracker для зберігання стану синхронізації.
3. Реалізація pipeline інкрементального імпорту.
4. Інтеграція виявлення видалених позицій та захисту від подвійного запуску.
5. Тестування на повному обсязі каталогу (до 500 000 SKU) та моніторинг.

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

• Розробка state tracker для зберігання стану синхронізації
• Реалізація обраної стратегії визначення змін (timestamp, cursor, хеш, diff)
• Механізм виявлення та обробки видалених позицій
• Захист від подвійного запуску за допомогою лока
• Тестування на каталозі до 500 000 SKU
• Документація з запуску та моніторингу

Терміни реалізації

Точні терміни залежать від складності API постачальника та обсягу каталогу. Зв'яжіться з нами для оцінки вашого проекту — ми підготуємо індивідуальну пропозицію та покажемо, як інкрементальний імпорт товарів може скоротити ваші витрати на інфраструктуру до 80%. Замовте реалізацію інкрементального імпорту та отримайте безкоштовну консультацію з оптимізації синхронізації вашого каталогу.