Какие знаешь методы оптимизации CPU bound задач в Python?

Оптимизация CPU-bound задач в Python

CPU-bound задачи интенсивно используют процессор (математические вычисления, обработка данных, алгоритмы). Python медленнее других языков для таких задач из-за GIL (Global Interpreter Lock). Вот методы оптимизации.

1. Многопроцессность (Multiprocessing)

Один из лучших способов обойти GIL — использовать несколько процессов.

import multiprocessing
import time

def cpu_intensive_task(n):
    """Факториал - CPU-bound задача"""
    result = 1
    for i in range(2, n + 1):
        result *= i
    return result

# Однопоточно - медленно
start = time.time()
results = [cpu_intensive_task(100000) for _ in range(4)]
print(f'Однопоточный: {time.time() - start:.2f}s')

# Многопроцессный - быстро (обходит GIL)
if __name__ == '__main__':
    start = time.time()
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(cpu_intensive_task, [100000] * 4)
    print(f'Многопроцессный: {time.time() - start:.2f}s')
    # Результат: 2-3x ускорение на 4-ядерной системе

Преимущества:

• Обходит GIL
• True параллелизм

Недостатки:

• Большие накладные расходы на создание процессов
• Сложнее обмениваться данными между процессами

2. Cython — компиляция в C

Компилируешь Python в C код для 10-100x ускорения.

# fibonacci.pyx (Cython файл)
cdef int fib(int n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

def fibonacci(int n):
    return fib(n)

# Компиляция
# python setup.py build_ext --inplace

# Использование
# from fibonacci import fibonacci
# result = fibonacci(35)  # Очень быстро!

setup.py:

from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize('fibonacci.pyx')
)

Результат: Python fib(35) = 10 секунд, Cython = 0.1 секунд

3. NumPy для векторизации

Векторизованные операции работают на скомпилированном C коде.

import numpy as np
import time

# Медленное решение - цикл Python
def python_sum(n):
    result = 0
    for i in range(n):
        result += i
    return result

# Быстрое решение - NumPy
def numpy_sum(n):
    return np.sum(np.arange(n))

# Сравнение
n = 10**7
start = time.time()
python_result = python_sum(n)
print(f'Python: {time.time() - start:.3f}s')

start = time.time()
numpy_result = numpy_sum(n)
print(f'NumPy: {time.time() - start:.3f}s')
# NumPy в 100+ раз быстрее

Пример: матричные операции

import numpy as np

# Решение системы линейных уравнений
A = np.random.rand(1000, 1000)
b = np.random.rand(1000)

# Очень быстро через NumPy (использует LAPACK)
x = np.linalg.solve(A, b)

# Перемножение матриц
result = A @ A.T  # Использует BLAS — оптимизированная библиотека

4. PyPy — альтернативный интерпретатор

PyPy использует JIT компиляцию для значительного ускорения.

# Установка PyPy
# conda install -c conda-forge pypy
# pypy3 script.py

# Пример скрипта
import time

def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

start = time.time()
result = fibonacci(35)
print(f'CPython: ~10 сек')
print(f'PyPy: ~0.1 сек')

Преимущества PyPy:

• 5-10x ускорение для CPU-bound кода
• Прозрачное использование (просто замени интерпретатор)

Недостатки:

• Не совместим со всеми библиотеками (особенно C extensions)
• Первый запуск медленнее (JIT компиляция)

5. numba — JIT компиляция для NumPy

Векторизуй NumPy функции с помощью @jit декоратора.

from numba import jit
import numpy as np
import time

# Обычная Python функция
def calculate_mandelbrot(xmin, xmax, ymin, ymax, width, height, max_iter):
    result = np.zeros((height, width))
    
    for y in range(height):
        for x in range(width):
            c = complex(xmin + (x/width)*(xmax-xmin), 
                       ymin + (y/height)*(ymax-ymin))
            z = 0
            for i in range(max_iter):
                if abs(z) > 2:
                    result[y, x] = i
                    break
                z = z*z + c
    return result

# С @jit — скомпилируется в machine code
@jit(nopython=True)
def calculate_mandelbrot_jit(xmin, xmax, ymin, ymax, width, height, max_iter):
    result = np.zeros((height, width))
    
    for y in range(height):
        for x in range(width):
            c = complex(xmin + (x/width)*(xmax-xmin), 
                       ymin + (y/height)*(ymax-ymin))
            z = 0
            for i in range(max_iter):
                if abs(z) > 2:
                    result[y, x] = i
                    break
                z = z*z + c
    return result

# Сравнение
start = time.time()
result1 = calculate_mandelbrot(-2.5, 1, -1.25, 1.25, 512, 512, 100)
print(f'Python: {time.time() - start:.2f}s')

start = time.time()
result2 = calculate_mandelbrot_jit(-2.5, 1, -1.25, 1.25, 512, 512, 100)
print(f'Numba JIT: {time.time() - start:.2f}s')
# Numba: 50-100x ускорение

6. Параллельные вычисления с joblib

Проще чем multiprocessing, встроена поддержка кэширования.

from joblib import Parallel, delayed
import time

def expensive_function(x):
    total = 0
    for i in range(10**7):
        total += i * x
    return total

# Однопоточно
start = time.time()
results = [expensive_function(x) for x in range(4)]
print(f'Последовательно: {time.time() - start:.2f}s')

# Параллельно
start = time.time()
results = Parallel(n_jobs=4)(delayed(expensive_function)(x) for x in range(4))
print(f'Параллельно: {time.time() - start:.2f}s')
# Почти 4x ускорение

7. Оптимизация алгоритмов

Иногда лучший способ — улучшить сам алгоритм.

# Плохо: O(n²) сложность
def slow_sum_pairs(arr):
    count = 0
    for i in range(len(arr)):
        for j in range(len(arr)):
            if arr[i] + arr[j] == target:
                count += 1
    return count

# Хорошо: O(n) с хеш-таблицей
def fast_sum_pairs(arr, target):
    seen = {}
    count = 0
    for num in arr:
        complement = target - num
        if complement in seen:
            count += seen[complement]
        seen[num] = seen.get(num, 0) + 1
    return count

# Рекурсия vs итерация
def slow_fibonacci(n):
    if n < 2:
        return n
    return slow_fibonacci(n-1) + slow_fibonacci(n-2)  # O(2^n)

def fast_fibonacci(n):
    if n < 2:
        return n
    a, b = 0, 1
    for _ in range(n-1):
        a, b = b, a + b
    return b  # O(n)

# Результат: fib(35) медленно 10с, быстро 0.000001с

8. Профилирование для определения bottleneck

Найди узкие места перед оптимизацией.

import cProfile
import pstats

def test_function():
    # CPU-bound код
    total = 0
    for i in range(10**6):
        total += i
    return total

# Профилирование
profile = cProfile.Profile()
profile.enable()
test_function()
profile.disable()

stats = pstats.Stats(profile)
stats.sort_stats('cumulative')
stats.print_stats(10)  # Top 10 функций

# Альтернатива: line_profiler
# pip install line_profiler
# kernprof -l -v script.py

9. Сравнительная таблица методов

Практический пример: обработка изображений

import numpy as np
from numba import jit
import cv2

# Медленный Python
def apply_filter_slow(image):
    result = np.zeros_like(image)
    for y in range(1, image.shape[0]-1):
        for x in range(1, image.shape[1]-1):
            # Применяем фильтр
            result[y, x] = np.sum(image[y-1:y+2, x-1:x+2]) / 9
    return result

# С Numba JIT
@jit(nopython=True)
def apply_filter_fast(image):
    result = np.zeros_like(image, dtype=np.float32)
    for y in range(1, image.shape[0]-1):
        for x in range(1, image.shape[1]-1):
            total = 0.0
            for i in range(-1, 2):
                for j in range(-1, 2):
                    total += image[y+i, x+j]
            result[y, x] = total / 9.0
    return result

# Или используй NumPy vectorization
def apply_filter_numpy(image):
    # Свёртка - оптимизирована на C
    kernel = np.ones((3, 3)) / 9
    return cv2.filter2D(image, -1, kernel)

Выводы

Стратегия оптимизации CPU-bound:

1. Первый шаг: улучши алгоритм (часто дает 10-1000x)
2. Второй шаг: используй NumPy для векторизации
3. Третий шаг: примени Numba @jit для критичного кода
4. Четвёртый шаг: используй multiprocessing если нужен параллелизм
5. Пятый шаг: переписывай на Cython только если очень нужно

Никогда не оптимизируй рано — профилируй first!

Помни: GIL в Python означает что threading не помогает для CPU-bound, но multiprocessing работает отлично.