Слить два журнала Python по времени: heapq.merge без общей сортировки

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

Потребовать порядок каждого входа

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

Главное предусловие находится у источников. merge не сортирует их за вас. Запись с более ранним временем, спрятанная после поздней в одном файле, станет видна слишком поздно. Это особенно важно для журналов из нескольких потоков: порядок физической записи не всегда совпадает с временной меткой события. Перед выбором алгоритма проверьте, какой именно порядок гарантирует производитель.

Пример объединяет независимые источники A и B с равными временными метками. Для полного порядка используется составной ключ: время, затем идентификатор источника. Оба входа отсортированы по этому же ключу. Явный второй компонент делает результат понятным при равенстве времени и избавляет отчёт от случайной зависимости от порядка переданных аргументов.

Слить записи по составному ключу

from heapq import merge
from operator import itemgetter

left = iter([(1, "A", "start"), (4, "A", "done")])
right = iter([(2, "B", "start"), (4, "B", "done")])
key = itemgetter(0, 1)
result = list(merge(left, right, key=key))
assert result == [
    (1, "A", "start"),
    (2, "B", "start"),
    (4, "A", "done"),
    (4, "B", "done"),
]
assert [key(row) for row in result] == sorted(key(row) for row in result)
assert len(result) == 4
print([(time, source) for time, source, _ in result])

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

Когда поток можно выдавать дальше

Если входы отсортированы по убыванию, параметр reverse=True подходит только тогда, когда каждый из них уже имеет убывающий порядок. Он не разворачивает источники. Для файлового генератора разворот вообще может потребовать предварительного чтения или специального доступа. Сначала договоритесь о направлении сортировки в формате выгрузки, затем передавайте его алгоритму слияния.

Количество источников тоже влияет на решение. merge хорошо подходит к известному набору отсортированных потоков. Для живых сетевых событий отсутствие следующего значения у одного источника создаёт вопрос: можно ли уже выдать запись из другого или позднее придёт более ранняя? Это задача о допустимом опоздании и границе завершённого времени. Обычный синхронный merge её не решает; вход должен иметь понятную семантику следующего элемента и конца.

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

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

Продолжить практику

В «Python для профессионалов» вы продолжите работу с итераторами, генераторами и стандартной библиотекой. Откройте программу и переходите к конвейерам, где источник, порядок и момент чтения заданы явно: это позволяет обрабатывать большие входы постепенно и понимать стоимость каждого шага.