predict по потоку: ограничиваем размер порции в памяти

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

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

Большинство estimator scikit-learn ожидают готовую двумерную порцию признаков. Им не следует передавать произвольный генератор в надежде на потоковую обработку. Отдельная функция читает до batch_size строк через islice, вызывает predict и выдаёт результаты. Когда источник закончился, пустая порция завершает цикл без вызова модели.

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

Проверить реальные границы вызовов

Модель обучается один раз до чтения потока. Функция counted_predict записывает размеры входов и делегирует настоящему LinearRegression. Это позволяет проверить, что семи строкам соответствуют вызовы по три, три и одной строке. Вспомогательная функция не подменяет прогноз константой.

Отдельный вызов с пустым источником должен вернуть пустой результат и не обращаться к estimator. Многие модели отклоняют predict с нулём строк. Проверка пустоты нужна до вызова, а не в обработчике общей ошибки после него.

from itertools import islice
from math import isclose
from sklearn.linear_model import LinearRegression

def predict_stream(rows, predict, batch_size):
    if batch_size <= 0:
        raise ValueError("batch_size must be positive")
    iterator = iter(rows)
    while True:
        batch = list(islice(iterator, batch_size))
        if not batch:
            return
        yield from predict(batch)

model = LinearRegression().fit([[0], [1], [2]], [1, 3, 5])
calls = []
def counted_predict(batch):
    calls.append(len(batch))
    return model.predict(batch)

result = list(predict_stream(([i] for i in range(7)), counted_predict, 3))
assert calls == [3, 3, 1]
assert all(isclose(value, 2 * i + 1, abs_tol=1e-9)
           for i, value in enumerate(result))
assert list(predict_stream(iter(()), counted_predict, 3)) == []
assert calls == [3, 3, 1]
print("batch sizes:", calls, "predictions:", result)

Выбирать размер по рабочему пределу

Размер порции влияет на память и накладные расходы вызова. Начните с небольшого безопасного значения, затем измерьте память и время на своей модели. Для алгоритмов с большим промежуточным расчётом, например поиском расстояний, расход может расти быстрее числа входных строк. Один универсальный batch_size для всех estimator не нужен.

Здесь один исполнитель сохраняет порядок входа. При параллельной обработке или фильтрации дополнительно сохраняйте идентификаторы. При сбое после выдачи нескольких порций потребуется отдельный протокол повторного расчёта: ограниченная память сама не обеспечивает возобновление и целостную публикацию результата.

Закрепить базовую операцию

В открытой практике fit и predict сначала получите прогноз для готового небольшого набора. Затем оберните predict ограничением порции и проверьте пустой источник. Откройте урок «Первый fit и predict на своих данных». Затем вернитесь к сценарию статьи и проверьте его на своём небольшом наборе. Посмотрите программу курса первой ML-модели, чтобы пройти путь от обучения к использованию результата.