Ray для распределенных вычислений: когда Python начинает бежать, а не ползать

Python медленный. Все это знают

Вы пишете скрипт для обработки данных. Запускаете. Ждете. Пьете кофе. Проверяете почту. Возвращаетесь — все еще ждете. Знакомая история?

Встроенный multiprocessing в Python — это как пытаться собрать мебель IKEA голыми руками. Теоретически можно. Практически — больно, медленно и половина деталей оказывается лишней.

Зачем Ray, если есть multiprocessing?

Хороший вопрос. Multiprocessing — это как советская "Жигули". Едет? Едет. Удобно? Нет. Надежно? Ха-ха.

Вот что Ray делает лучше:

• Автоматически распределяет задачи между ядрами процессора
• Умеет работать с состоянием (stateful workers) — в multiprocessing это головная боль
• Масштабируется на кластер из машин — сегодня на локальном ПК, завтра на 100 серверах
• Имеет встроенную отказоустойчивость
• Поддерживает GPU вычисления (полезно для запуска LLM на нескольких видеокартах)

Сначала покажу, как НЕ надо делать

Классический подсчет простых чисел в одном потоке:

import time

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

def count_primes_sequential(limit):
    count = 0
    for num in range(2, limit):
        if is_prime(num):
            count += 1
    return count

if __name__ == "__main__":
    start = time.time()
    result = count_primes_sequential(100000)
    elapsed = time.time() - start
    print(f"Найдено простых чисел: {result}")
    print(f"Время выполнения: {elapsed:.2f} секунд")

На моем ноутбуке с i7 это занимает примерно 3.2 секунды. Медленно? Да. Но главное — процессор загружен только на 13% (одно ядро из восьми). Остальные семь простаивают.

Теперь покажу, как должно быть

Устанавливаем Ray одной командой:

pip install ray

Вот так выглядит параллельная версия:

import ray
import time
import numpy as np

# Инициализируем Ray
ray.init(num_cpus=8)  # Используем все 8 ядер

@ray.remote
def is_prime_ray(n):
    if n < 2:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

@ray.remote
def count_primes_chunk(start, end):
    count = 0
    for num in range(start, end):
        if is_prime_ray.remote(num):
            count += 1
    return count

def count_primes_parallel(limit, num_chunks=8):
    # Разбиваем диапазон на части
    chunk_size = limit // num_chunks
    chunks = [(i * chunk_size, (i + 1) * chunk_size) 
              for i in range(num_chunks)]
    
    # Запускаем все задачи параллельно
    futures = [count_primes_chunk.remote(start, end) 
               for start, end in chunks]
    
    # Ждем завершения всех задач
    results = ray.get(futures)
    
    # Суммируем результаты
    return sum(results)

if __name__ == "__main__":
    start = time.time()
    result = count_primes_parallel(100000)
    elapsed = time.time() - start
    print(f"Найдено простых чисел: {result}")
    print(f"Время выполнения: {elapsed:.2f} секунд")
    
    # Не забываем закрыть Ray
    ray.shutdown()

Результат? 0.8 секунды вместо 3.2. В 4 раза быстрее. И это на том же железе.

1 Почему это работает быстрее?

Ray создает отдельные процессы-работники (workers). Каждый работник получает свою порцию данных. Все работники работают одновременно. Когда все закончили — результаты собираются.

2 А что с накладными расходами?

Есть. Создание процессов, передача данных между ними — это время. Но на задачах, которые выполняются дольше 100 миллисекунд, накладные расходы окупаются с лихвой. Для микро-задач лучше использовать threading или asyncio.

"Но у меня не только простые числа!"

Понимаю. Вот реальные сценарии, где Ray спасает:

А что с альтернативами?

Есть их много. Все плохие по-своему.

• multiprocessing.Pool — базовый вариант. Работает, но масштабирование на несколько машин не предусмотрено. Как только задачи усложняются — начинаются проблемы.
• Dask — хорош для pandas/numpy, но для произвольного Python-кода слишком тяжеловесный.
• Celery — для фоновых задач, а не для высокопроизводительных вычислений. Очереди Redis, брокеры сообщений — это лишние сложности.
• Joblib — простой, но только для embarrassingly parallel задач (где задачи независимы).

Ray занимает золотую середину: достаточно прост для начала, достаточно мощный для сложных проектов.

Где Ray не поможет (и даже навредит)

Не все задачи можно параллелить. Если ваша задача:

1. Выполняется быстрее 50 мс — накладные расходы съедят всю выгоду
2. Требует постоянной синхронизации между процессами (общая память, блокировки)
3. Уже ограничена не CPU, а диском или сетью (например, скачивание одного большого файла)
4. Слишком простая для таких сложностей (print("Hello World") в 8 процессах — это уже психическое расстройство)

Практический совет: начните с малого

Не переписывайте весь проект под Ray сразу. Выделите самый медленный кусок кода. Обычно это:

• Цикл for, который обрабатывает тысячи элементов
• Применение одной функции к списку данных
• Обучение нескольких моделей с разными гиперпараметрами

Возьмите этот кусок, оберните в @ray.remote функцию. Запустите параллельно. Измерьте разницу.

# Было
results = []
for item in large_list:
    result = expensive_function(item)
    results.append(result)

# Стало
@ray.remote
def process_item(item):
    return expensive_function(item)

futures = [process_item.remote(item) for item in large_list]
results = ray.get(futures)

Иногда этого достаточно для ускорения в 5-10 раз.

А что насчет GPU?

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

Но будьте готовы к настройке. Не всегда получается "включить и работать". Иногда приходится бороться с драйверами, CUDA, памятью. Как в случае с GRPO + LoRA на нескольких GPU — там каждый мегабайт VRAM на счету.

Ошибки, которые все совершают (и вы тоже)

1. Слишком мелкие задачи — отправлять в Ray функцию, которая выполняется 1 мс, бессмысленно. Группируйте задачи в батчи.
2. Забывают ray.shutdown() — процессы остаются висеть в памяти. Через день обнаруживаете 50 питоновых процессов.
3. Передают большие объекты между процессами — сериализация/десериализация занимает время. Если возможно, создавайте данные внутри worker.
4. Не учитывают ограничения памяти — 8 процессов, каждый с копией датасета на 2 ГБ = 16 ГБ оперативки. Убедитесь, что хватит.

Что дальше? Из локального ПК в облако

Самое интересное в Ray — что тот же код, который работает на вашем ноутбуке, может работать на кластере из 100 машин в AWS или GCP. Меняется только инициализация:

# Вместо ray.init() на локальной машине
ray.init(address="auto")  # Подключение к существующему кластеру

# Или при запуске из командной строки
# ray start --head  # на головной ноде
# ray start --address='head_node_ip:6379'  # на рабочих нодах

Это меняет правила игры. Можно прототипировать на локальной машине, а когда нужно обработать терабайты данных — развернуть на кластере без переписывания кода.

Итог: кому нужен Ray?

Вам, если:

• Python-скрипты работают дольше минуты
• Процессор загружен меньше чем на 30% при этом
• Есть задачи типа "применить функцию к списку из 10 000 элементов"
• Планируете масштабироваться на несколько машин (даже если не сейчас)
• Работаете с ML/AI и хотите ускорить preprocessing или hyperparameter tuning

Не тратьте время, если:

• Код и так выполняется за секунды
• У вас одноядерный процессор (такие еще есть?)
• Задачи сильно связаны между собой (много синхронизации)
• Нет времени разбираться с новой технологией (хотя Ray довольно прост)

Попробуйте на самом медленном куске вашего кода. Если не ускорится в 2-3 раза — значит, задача не для параллелизации. Если ускорится — теперь вы знаете инструмент, который сэкономит вам часы ожидания.

И помните: быстрый код — это не только про экономию времени. Это про возможность экспериментировать быстрее, тестировать больше гипотез, итерации сокращаются с часов до минут. А это в разработке важнее, чем кажется.