Реализация инкрементального импорта товаров (только изменения)

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

Мы решаем эту проблему с помощью инкрементального импорта: обрабатываем только изменённые позиции. За несколько лет мы внедрили такую схему для 15 проектов — от небольших магазинов до маркетплейсов с каталогом до 500 000 SKU. Наш опыт гарантирует, что вы получите работающее решение в сжатые сроки.

Инкрементальный импорт (delta import) — это метод синхронизации, при котором обновляются только изменившиеся записи. Это снижает нагрузку на сервер на 80% и сокращает время импорта в 10 раз по сравнению с полной перезагрузкой. Например, на одном из проектов с каталогом 200 000 SKU мы сократили время синхронизации с 3 часов до 18 минут, используя timestamp и хэш-сравнение для надёжности.

Как выбрать стратегию определения изменений?

Выбор метода зависит от возможностей поставщика данных. Сравним основные подходы в таблице:

Рассмотрим каждый метод подробнее.

Временная метка

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

Курсор / change log

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

Хэш-сравнение

Отметим: когда источник не поддерживает фильтрацию по изменениям — сравниваем хэш строки:

$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 час, этого достаточно для большинства каталогов. Это предотвращает дублирование и конфликты.

Что входит в работу?

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

Сроки реализации

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