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

Представьте: вы управляете интернет-магазином с каталогом 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%. Закажите реализацию инкрементального импорта и получите консультацию по оптимизации синхронизации вашего каталога.