Генератор Python: как обработать большой учебный файл без списка всех строк

Если нужно один раз пройти по большому текстовому файлу и посчитать только подходящие строки, лишним шагом часто становится readlines(). Этот вызов создаёт список со всеми строками файла, хотя для подсчёта каждая строка нужна лишь на момент проверки. В таких задачах удобнее оставить файл потоком данных: строка читается, фильтруется и сразу перестаёт быть нужна.

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

Задача последовательной обработки

Файловый объект в Python уже умеет отдавать строки по одной в цикле for. Это подходит для последовательной обработки, когда итогом нужен счётчик, сумма, фильтрованный вывод или запись в другой файл. В таком сценарии не требуется доступ к произвольной строке по индексу и не нужно хранить весь набор в памяти одновременно.

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

Одинаковый подсчёт двумя способами

Учебный пример. Пусть есть условный файл students.log в UTF-8. Каждая непустая строка имеет вид student_id;group;status;score. Нужно посчитать строки, где group == "python", status == "done" и score >= 80. Для проверки результата возьмём блок из 10 строк, в котором ровно 3 строки подходят условию, и повторим его 100 000 раз. Тогда в файле будет 1 000 000 строк, а ожидаемый итог подсчёта составит 3 * 100000 = 300000.

BLOCK = [
    "s001;python;done;82\n",   # подходит
    "s002;python;draft;90\n",  # не подходит
    "s003;js;done;88\n",       # не подходит
    "s004;python;done;79\n",   # не подходит
    "s005;python;done;95\n",   # подходит
    "s006;python;done;bad\n",  # не подходит
    "s007;python;done;80\n",   # подходит
    "s008;python;skipped;91\n",# не подходит
    "\n",                      # пустая строка
    "s010;python;done;50\n"    # не подходит
]

REPEATS = 100_000

with open("students.log", "w", encoding="utf-8") as f:
    for _ in range(REPEATS):
        f.writelines(BLOCK)

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

def is_target(line: str) -> bool:
    parts = line.strip().split(";")
    if len(parts) != 4:
        return False
    _, group, status, score_text = parts
    if group != "python" or status != "done":
        return False
    try:
        score = int(score_text)
    except ValueError:
        return False
    return score >= 80


def count_with_list(path: str) -> int:
    with open(path, "r", encoding="utf-8") as f:
        lines = f.readlines()
    return sum(1 for line in lines if is_target(line))


def count_with_stream(path: str) -> int:
    with open(path, "r", encoding="utf-8") as f:
        return sum(1 for line in f if is_target(line))

Ожидаемый практический результат в этом учебном кейсе такой: обе функции должны вернуть 300000. Разница между ними не в математике подсчёта, а в том, что второй способ не обязан держать список из миллиона строк до завершения работы.

Итератор и исчерпание: где чаще всего ошибаются

Когда вы итерируетесь по файлу или по генераторному выражению, вы работаете не с контейнером, а с потоком элементов. Такой объект движется только вперёд. Если вы уже использовали его в sum(), for или list(), данные считаются потреблёнными. Это полезно для последовательной обработки, но неожиданно для тех, кто привык к спискам.

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

with open("students.log", "r", encoding="utf-8") as f:
    rows = (line for line in f if is_target(line))
    first_pass = sum(1 for _ in rows)
    second_pass = sum(1 for _ in rows)

print(first_pass)   # 300000
print(second_pass)  # 0

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

Как собрать конвейер строк без промежуточного списка

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

Ниже тот же подсчёт оформлен как цепочка генераторов. Логика не стала «быстрее по определению», но стала прозрачнее: отдельно видно очистку строк, отдельно валидацию структуры и отдельно фильтр целевых записей.

def cleaned_lines(file_obj):
    for raw in file_obj:
        line = raw.strip()
        if line:
            yield line


def valid_rows(lines):
    for line in lines:
        parts = line.split(";")
        if len(parts) == 4:
            yield parts


def target_rows(rows):
    for student_id, group, status, score_text in rows:
        if group != "python" or status != "done":
            continue
        try:
            score = int(score_text)
        except ValueError:
            continue
        if score >= 80:
            yield student_id, score


def count_with_pipeline(path: str) -> int:
    with open(path, "r", encoding="utf-8") as f:
        rows = valid_rows(cleaned_lines(f))
        return sum(1 for _ in target_rows(rows))

Такой вариант полезен не только для чтения больших файлов. Он помогает локально тестировать этапы: например, можно отдельно проверить, как valid_rows() ведёт себя на битой строке, не поднимая весь сценарий целиком. Если позже понадобится не только число строк, но и сумма баллов, вы замените только финальный шаг, не переписывая всю цепочку.

Как проверить результат и память без догадок

Удачный переход на генераторы проверяют не по ощущению «стало легче», а по двум критериям. Первый — совпал ли результат подсчёта. Второй — изменился ли пик памяти в вашем сценарии. Если числа разные, проблема не в генераторах как идее, а в отличии логики фильтрации, порядке очистки строк или обработке ошибок преобразования.

Для учебного сравнения можно использовать tracemalloc. Он показывает объём памяти, который был выделен Python во время выполнения конкретной функции. Это не полный снимок всей памяти процесса и не универсальный бенчмарк, но для сравнения двух реализаций одного и того же подсчёта в одном скрипте этого достаточно.

import tracemalloc


def measure(func, path: str):
    tracemalloc.start()
    result = func(path)
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return result, peak


path = "students.log"
list_result, list_peak = measure(count_with_list, path)
stream_result, stream_peak = measure(count_with_stream, path)
pipeline_result, pipeline_peak = measure(count_with_pipeline, path)

print(list_result, list_peak)
print(stream_result, stream_peak)
print(pipeline_result, pipeline_peak)

Что считать хорошим итогом в этом кейсе: list_result == stream_result == pipeline_result == 300000, а пиковое значение у потоковых вариантов обычно ниже, чем у версии с readlines(). Конкретные байты зависят от содержимого файла, версии Python и окружения, поэтому их лучше измерять у себя, а не брать из чужого примера.

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

Связанная тема для дальнейшей практики

Во внутренней карточке каталога курса «Fullstack-разработчик на Python + ИИ» сохранено описание с темами по лямбда-функциям, итераторам, генераторам, обработке исключений и регулярным выражениям. На официальной странице школы курс описан шире, как программа по веб-разработке на Python; для нашей темы полезно сверить у себя, достаточно ли вам короткого блока по генераторам или нужна практика с файлами, исключениями и проверкой решений.