Повторный запуск миграции создает дубли: как сделать ETL идемпотентным

Миграцию приходится повторять после сетевого сбоя, таймаута или исправления преобразования. Если повторный запуск создает новые копии записей, процесс не умеет отличать уже обработанный объект от нового и не является идемпотентным.

Основа исправления — стабильный ключ источника, уникальное ограничение в приемнике и операция upsert. Checkpoint помогает продолжать работу, но не заменяет уникальность: задача может завершиться после записи и до отметки прогресса.

Что проверить в первую очередь

Сначала зафиксируйте наблюдаемое поведение и не меняйте сразу несколько настроек. Важны точное время сбоя, адрес или сценарий, ожидаемый результат и последнее известное рабочее состояние. Так можно отличить причину от случайного совпадения и сохранить возможность быстрого отката.

• Определите, есть ли у каждой исходной сущности неизменяемый уникальный ID.
• Найдите, по какому набору полей сейчас решается, новая запись или существующая.
• Проверьте уникальные индексы приемника и поведение при конфликте.
• Воспроизведите сбой после записи, но до сохранения checkpoint.

Почему возникает проблема

У подобных сбоев редко бывает одна универсальная причина. На итог одновременно влияют конфигурация приложения, окружение, данные, кеш, права и внешние сервисы. Проверка должна идти от внешнего симптома к конкретному уровню, на котором впервые появляется неверное состояние.

• Каждый запуск генерирует новый UUID вместо сохранения связи source_id → target_id.
• Поиск дубля выполняется по изменяемым полям: имени, email или времени импорта.
• Проверка SELECT затем INSERT не защищена от параллельных воркеров.
• Checkpoint обновляется отдельно и теряется после частичного сбоя.
• Одна сущность приходит из нескольких источников без namespace или приоритета.

Пошаговая диагностика

Диагностику проводите на копии или в контролируемое время. Перед изменениями сохраните конфигурацию, данные и журналы. Каждый шаг должен отвечать на один вопрос и оставлять измеримый результат: код ответа, запись в логе, состояние процесса, значение поля или воспроизводимый тест.

• Выберите один дубликат и сравните source ID, payload, время вставки и запуск задачи.
• Проверьте план и ограничения таблицы, а не только прикладную проверку exists.
• Запустите один небольшой batch дважды и сравните количество insert/update/no-op.
• Искусственно оборвите задачу между пакетами и повторите ее на тестовой базе.
• Проверьте параллельный запуск двух воркеров на пересекающемся диапазоне.

Идемпотентность на уровне данных

Надежность обеспечивается не флагом задачи, а инвариантом, который база может проверить атомарно.

• Храните source_system и source_id с уникальным составным индексом.
• Используйте upsert с явным перечнем обновляемых полей и правилами разрешения конфликтов.
• Для событий применяйте уникальный event_id и отдельный журнал обработки.
• Checkpoint сохраняйте после коммита batch, но допускайте безопасное повторение последнего пакета.
• Для удалений определите tombstone/статус, чтобы повтор не воскресил устаревшую запись.

Как исправить проблему

Исправление лучше разбить на небольшие обратимые изменения. После каждого шага повторяйте исходный сценарий и проверяйте соседние функции. Если правка касается данных, сначала выполните ее на ограниченной выборке и сохраните журнал затронутых записей.

• Добавьте стабильный внешний ключ и заполните его для существующих записей после сверки.
• Создайте unique constraint, предварительно подготовив отчет о найденных конфликтах.
• Замените SELECT+INSERT атомарным upsert или транзакцией с блокировкой.
• Разбейте миграцию на ограниченные batch с журналом результата и повторов.
• После выполнения запустите reconciliation: количество, суммы, связи и выборочные хеши.

Безопасный порядок внедрения

• Сохраните резервную копию затрагиваемых файлов, базы и конфигурации, а также заранее опишите способ отката.
• Воспроизведите сбой на тестовой записи, учетной записи или отдельном окружении без реальных платежей и рассылок.
• Вносите по одному логическому изменению, фиксируя его в системе контроля версий или журнале работ.
• Не отключайте права, проверку входных данных и защитные механизмы только ради исчезновения сообщения об ошибке.
• После выкладки контролируйте логи, метрики и ключевой пользовательский сценарий, а не только открытие одной страницы.

Как проверить результат

Успешный разовый тест еще не доказывает исправление. Нужны повторный запуск, крайние случаи и проверка после очистки кеша, перезапуска процесса или новой сессии. Для критичных сценариев полезно сохранить автоматический тест или хотя бы короткий регрессионный чек-лист.

• Два одинаковых запуска не увеличивают количество целевых сущностей.
• Повтор после искусственного обрыва корректно обновляет или пропускает последний batch.
• Параллельные воркеры не создают дубли благодаря ограничению базы.
• Отчет разделяет inserted, updated, skipped и conflicted и сходится с источником.

Типичные ошибки при исправлении

• Удалять дубли до добавления правила, которое не даст им появиться снова.
• Считать email универсальным стабильным ID для всех сущностей.
• Полагаться только на checkpoint без unique constraint.
• Применять upsert, который незаметно перетирает более новые данные старым источником.

Как предотвратить повторение

Профилактика строится вокруг наблюдаемости и воспроизводимости: понятных конфигураций, контролируемых релизов, журналов без секретов и тестов на реальные сценарии. Важно не просто убрать текущий симптом, а сделать следующий похожий сбой заметным раньше пользователя.

• Проектируйте миграции как повторяемые операции с самого начала.
• Добавляйте тест двойного запуска и аварийного продолжения.
• Храните версию преобразования и идентификатор batch у измененных записей.
• Проверяйте инварианты и reconciliation автоматически после каждого запуска.

Что подготовить для разбора

• Краткое описание ожидаемого и фактического поведения без паролей, токенов и персональных данных.
• Точное время и последовательность действий, после которых появляется проблема.
• Версии приложения, окружения и зависимостей, а также перечень последних изменений.
• Фрагменты журналов с контекстом до и после ошибки; секреты в них необходимо скрыть.
• Описание уже выполненных проверок и способ безопасно повторить проблему.

Частые вопросы

Можно ли просто пропускать уже существующий email?

Это опасно, если email меняется или принадлежит разным типам сущностей. Лучше использовать неизменяемый ID источника и namespace.

Нужен ли checkpoint при наличии upsert?

Да, он повышает эффективность и позволяет продолжать с нужного места, но корректность при повторе обеспечивает идемпотентная запись и ограничения.

Когда нужна помощь специалиста

Если миграция создает дубли или ее страшно перезапускать, я могу разобрать ключи и точки отказа, сделать загрузку идемпотентной, подготовить безопасную очистку существующих дублей и отчет сверки.