ETL пропускает строки с одинаковым временем: составной курсор PostgreSQL

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

Время само по себе не задаёт уникальный порядок. Добавим устойчивый id и будем сравнивать пару: сначала время, затем идентификатор. ORDER BY и условие продолжения должны использовать те же поля в том же порядке. Оба поля должны быть непустыми, а пара — однозначно определять позицию.

Проверим границу на маленькой выгрузке

В примере четыре записи, три из них имеют одинаковое время. Размер страницы равен двум, поэтому граница намеренно проходит внутри группы. Python воспроизводит лексикографическое сравнение пары; ниже объясняется соответствующий запрос PostgreSQL.

rows = [
    ("2026-11-03T10:00:00Z", 11, "A"),
    ("2026-11-03T10:00:00Z", 12, "B"),
    ("2026-11-03T10:00:00Z", 13, "C"),
    ("2026-11-03T10:01:00Z", 14, "D"),
]
cursor = ("2026-11-03T09:00:00Z", 0)
seen = []
while True:
    page = [r for r in sorted(rows) if r[:2] > cursor][:2]
    if not page:
        break
    seen.extend(r[1] for r in page)
    cursor = page[-1][:2]
    print("page", [r[1] for r in page], "cursor", cursor)
assert seen == [11, 12, 13, 14]

Первая страница содержит 11 и 12, вторая — 13 и 14. Для PostgreSQL запрос имеет форму SELECT id, updated_at, payload FROM events WHERE (updated_at, id) > (%s, %s) ORDER BY updated_at, id LIMIT %s. Значения передаются параметрами драйвера, время — типизированным timestamp. Индекс по (updated_at, id) помогает обслуживать такое продолжение.

Пример использует одинаковые UTC-строки фиксированного формата только для наглядной локальной проверки. В базе храните время подходящим типом и не сравнивайте произвольные строки с разными смещениями. id должен иметь одинаковый тип в условии и сортировке.

Курсор фиксирует завершённый результат

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

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

Выберите договор с источником

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

Проверяйте две вещи отдельно: все ли строки статического снимка прочитаны и как источник ведёт себя при параллельном изменении. Тест на одинаковые timestamps ловит первую ошибку, а сценарий с задержанной транзакцией — вторую. Это помогает не приписывать одному SQL-условию гарантии всего конвейера.

В курсе Data Engineering Fast-Track вы соберёте конвейер Kafka → Spark → Lakehouse под управлением Airflow и отработаете повторные запуски и backfill. Откройте программу, чтобы связать чтение источника, сохранение результата и восстановление после сбоя.