Привет, Хабр! Сегодня разберемся с одной важной темой, которая может серьезно улучшить производительность Python‑кода — параллельные вычисления с помощью Joblib.

Joblib — это Python‑библиотека, которая предоставляет инструменты для параллельных вычислений, кэширования и эффективной обработки данных. Она используется для ускорения выполнения операций, таких как многократные вычисления, обработка больших массивов данных и параллельная обработка однотипных задач.

Почему стоит использовать Joblib вместо стандартного multiprocessing?

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

  • Меньше накладных расходов: для некоторых операций Joblib может быть быстрее, чем использование дефолтных механизмов Python (например, multiprocessing).

  • Меньше кода: Joblib позволяет просто распараллелить вычисления с минимальными усилиями.

Параллельные вычисления с Joblib

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

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

from math import sqrt

data = [i for i in range(1000000)]

result = [sqrt(i ** 2) for i in data]

Это нормально, но можно ли ускорить процесс? С помощью Joblib можно распараллелить этот цикл.

Вот как это выглядит:

from joblib import Parallel, delayed
from math import sqrt

# Данные
data = [i for i in range(1000000)]

# Параллельная обработка
result = Parallel(n_jobs=4)(delayed(sqrt)(i ** 2) for i in data)
print(result[:10])  # Печатаем первые 10 результатов

  • n_jobs=4: говорим, что хотим использовать 4 ядра процессора для выполнения задачи параллельно.

  • delayed(sqrt): функция delayed просто оборачивает функцию, чтобы её можно было передавать в параллельные вычисления.

Это распараллеливает вычисление и ускоряет его.

Важно помнить, что Joblib использует множество процессов, и каждый процесс работает в своем собственном пространстве памяти. Это важно учитывать, когда у вас есть ограничения по памяти. Если задача использует много памяти, распараллеливание может привести к увеличению нагрузки на систему, а в некоторых случаях даже к её замедлению.

Использование потоков или процессов?

Мы рассмотрели использование процессов, но что, если ваша задача не связана с тяжелыми вычислениями, а, например, работает с I/O или сетевыми запросами? В таком случае использование потоков будет более подходящим, т.к Python позволяет эффективно работать с потоками в таких сценариях.

Пример с потоками:

from joblib import Parallel, delayed
import time

# Пример задачи с I/O операциями
def fetch_data(i):
    time.sleep(1)
    return i

# Параллельная обработка с потоками
result = Parallel(n_jobs=4, prefer="threads")(delayed(fetch_data)(i) for i in range(10))
print(result)

Указываем prefer="threads", чтобы использовать потоки, а не процессы. Это важно, если задача не требует интенсивных вычислений, а скорее связана с ожиданием.

Работа с большими данными и меммапы

Когда данные не помещаются в память, необходимо использовать memory‑mapped files. Это позволяет работать с данными, не загружая их целиком в память. Joblib позволяет легко работать с такими данными.

Пример использования меммапов:

import numpy as np
from joblib import Parallel, delayed
import os
import tempfile

# Создаем временную директорию для хранения данных
temp_folder = tempfile.mkdtemp()

# Создаем массив данных
large_data = np.random.rand(int(1e8))  # Массив из 100 миллионов элементов

# Сохраняем данные с использованием меммапа
filename = os.path.join(temp_folder, "large_data.mmap")
np.save(filename, large_data)

# Загружаем данные через меммап
memmap_data = np.load(filename, mmap_mode='r')

# Параллельная обработка меммапа
result = Parallel(n_jobs=4)(delayed(np.sum)(memmap_data[i:i+10000]) for i in range(0, len(memmap_data), 10000))

print(result[:5])  # Печатаем первые несколько результатов

Здесь создаём массив с огромным количеством данных и используем меммап для того, чтобы не загружать его целиком в память.

Кэширование вычислений

Теперь поговорим о кэшировании. Когда задача повторяется, зачем снова вычислять одно и то же? Joblib позволяет легко кэшировать результаты вычислений, чтобы избежать повторной работы.

Пример кэширования:

from joblib import Memory

# Создаем кэш
memory = Memory("/tmp/joblib_cache", verbose=0)

@memory.cache
def expensive_computation(x):
    print(f"Выполняется сложная операция для {x}")
    return x ** 2

# Вызываем функцию
print(expensive_computation(10))
print(expensive_computation(10))  # Будет взят из кэша

Когда вы повторно вызываете expensive_computation(10), Joblib не будет повторно вычислять результат, а вернёт его из кэша. Это позволяет существенно сэкономить время, если у вас дорогостоящие вычисления.

А более подробно с библиотекой можно ознакомиться здесь.

Статья подготовлена в рамках специализации Python Developer, на которой можно прокачать навыки разработки на Python с нуля и до middle+ уровня. Ознакомиться с полной программой, а также посмотреть записи открытых уроков можно на странице специализации.