Профилирование и оптимизация
Правила оптимизации
«Преждевременная оптимизация — корень всех зол» — Дональд Кнут.
- Сначала правильность, потом скорость.
- Измеряйте, а не угадывайте. Узкое место почти всегда не там, где кажется.
- Оптимизируйте горячие места: 90% времени тратится в 10% кода.
- Алгоритм важнее микрооптимизаций: O(n²) → O(n) даёт больше, чем любые трюки.
- После изменения — измерьте снова и убедитесь, что тесты зелёные.
Замеры
import timeit
setup = "data = list(range(1000))"
print("цикл: ", timeit.timeit("r = []\nfor x in data: r.append(x * 2)", setup, number=2000))
print("включение: ", timeit.timeit("[x * 2 for x in data]", setup, number=2000))
print("map + lambda:", timeit.timeit("list(map(lambda x: x * 2, data))", setup, number=2000))
print("sum генератор:", timeit.timeit("sum(x for x in data)", setup, number=2000))
print("sum встроенная:", timeit.timeit("sum(data)", setup, number=2000))
timeit повторяет фрагмент много раз и отключает сборщик мусора — для микрозамеров. В консоли: python -m timeit -s "data=list(range(1000))" "sum(data)". В IPython/Jupyter — %timeit.
Для участков программы — time.perf_counter():
import time
from contextlib import contextmanager
@contextmanager
def measure(label):
start = time.perf_counter()
yield
print(f"{label}: {(time.perf_counter() - start) * 1000:.1f} мс")
with measure("сортировка 200 000 чисел"):
sorted(range(200_000, 0, -1))
cProfile — где тратится время
import cProfile, pstats, io
def parse(line):
return [int(x) for x in line.split(",")]
def validate(nums):
return all(n >= 0 for n in nums)
def slow_lookup(nums, allowed):
return [n for n in nums if n in allowed] # allowed — список!
def process(lines):
allowed = list(range(0, 1000, 3))
total = 0
for line in lines:
nums = parse(line)
if validate(nums):
total += len(slow_lookup(nums, allowed))
return total
lines = [",".join(str((i * j) % 1000) for j in range(20)) for i in range(800)]
profiler = cProfile.Profile()
profiler.enable()
process(lines)
profiler.disable()
out = io.StringIO()
pstats.Stats(profiler, stream=out).sort_stats("cumulative").print_stats(8)
print("\n".join(out.getvalue().splitlines()[:18]))
Колонки: ncalls — число вызовов, tottime — время в самой функции, cumtime — вместе с вызванными. Здесь видно, что время уходит в slow_lookup — поиск в списке. Замените allowed на множество и перезапустите пример.
Запуск для всей программы: python -m cProfile -s cumulative script.py или -o profile.out и просмотр в snakeviz.
Другие профилировщики
| Инструмент | Что показывает | Когда |
|---|---|---|
cProfile (stdlib) |
время по функциям | локально, первым делом |
line_profiler (@profile) |
время по строкам | когда функция найдена, но неясно, какая строка |
| py-spy | сэмплирование работающего процесса, flame graph | продакшен, без изменения кода |
scalene |
CPU + память + GPU, Python vs C | комплексный анализ |
tracemalloc, memray |
память | утечки, пиковое потребление |
| APM: Sentry, Datadog, OpenTelemetry | медленные запросы, SQL, внешние вызовы | веб-сервисы |
py-spy top --pid 12345 # «top» для Python-процесса
py-spy record -o flame.svg --pid 12345 # flame graph
Приёмы ускорения
1. Алгоритм и структура данных
Самый большой выигрыш. set для поиска, dict для индекса, deque для очереди, heapq для top-k, сортировка + бинарный поиск, предвычисление.
2. Встроенные функции и включения
Встроенные (sum, min, max, sorted, any, str.join, collections.Counter) написаны на C. Включения быстрее ручного append в цикле.
import timeit
setup = "words = ['кот', 'пёс', 'кот', 'ёж'] * 2500"
manual = """
counts = {}
for w in words:
if w in counts:
counts[w] += 1
else:
counts[w] = 1
"""
print("вручную:", round(timeit.timeit(manual, setup, number=50), 3))
print("Counter:", round(timeit.timeit("Counter(words)", "from collections import Counter\n" + setup, number=50), 3))
3. Кеширование
functools.cache для чистых функций, кеш результатов запросов (Redis), cached_property.
4. Ленивость
Генераторы вместо списков для потоковой обработки — меньше памяти, быстрее старт.
5. Векторизация
import numpy as np
prices = np.random.rand(10_000_000)
total = (prices * 1.2).sum() # одна операция на C над всем массивом — в 50–100 раз быстрее цикла
6. Ввод-вывод — главная причина медленных сервисов
- N+1 запросов: в цикле по 100 заказам делать запрос за клиентом → 101 запрос. Решение:
JOIN,select_related/prefetch_related(Django),selectinload(SQLAlchemy). - Батчи:
executemany,bulk_create, пакетные API. - Конкурентные сетевые запросы (asyncio, пул потоков).
- Индексы в базе данных, пагинация, кеширование.
7. Микрооптимизации (только в горячих циклах)
- локальная переменная вместо глобальной/атрибута (
append = result.append); __slots__для миллионов объектов;- избегать лишних вызовов функций и исключений в горячем цикле.
Когда этого мало
| Путь | Суть |
|---|---|
| PyPy | тот же код, JIT; хорошо для чистого Python |
| Cython / mypyc | компиляция типизированного Python в C-расширение |
| Rust + PyO3 / maturin | горячее ядро на Rust (так сделаны pydantic-core, ruff, polars) |
| numba | JIT для числовых функций с numpy |
| Процессы / очереди задач | распараллелить на ядра и машины |
Вопрос на собеседовании
«Сервис стал медленно отвечать. Ваши действия?» — Измерить: метрики и APM (какие эндпоинты, где время — БД, внешние API, CPU), профилирование (py-spy в проде, cProfile локально), логи медленных SQL. Типичные причины — N+1 запросов, отсутствие индексов, последовательные внешние вызовы, неудачная структура данных. Исправить узкое место, подтвердить замером, добавить тест/метрику против регрессии.
Итог
- Сначала измерить:
timeit,perf_counter,cProfile, py-spy. - Алгоритм и структура данных дают больше всего; затем встроенные функции, кеш, ленивость, векторизация.
- В веб-сервисах время чаще всего уходит на ввод-вывод: N+1, индексы, батчи, конкурентность.
- Микрооптимизации — только в доказанно горячих местах; дальше — PyPy, Cython, Rust.
Практика решено 0 из 5
Функция slow_fib(n) считает числа Фибоначчи экспоненциально долго.
Ускорьте её кешированием (functools.cache или lru_cache) без изменения рекурсивного алгоритма,
чтобы slow_fib(300) считалась мгновенно.
Какие тесты проверяют код (3)
assert hasattr(slow_fib, "cache_info"), "Добавьте декоратор functools.cache / lru_cache" assert slow_fib(10) == 55 assert hasattr(slow_fib, "cache_info") and slow_fib(300) == 222232244629420445529739893461909967206666939096499764990979600
from functools import cache
@cache
def slow_fib(n):
if n < 2:
return n
return slow_fib(n - 1) + slow_fib(n - 2)
С помощью cProfile и pstats найдите самую «дорогую» функцию.
Напишите hottest_function(func): запустить func() под профилировщиком и вернуть имя
функции (строку) с наибольшим собственным временем (tottime), исключая встроенные функции
(у них имя файла ~).
Какие тесты проверяют код (1)
def fast():
return sum(range(100))
def slow():
total = 0
for i in range(300_000):
total += i % 7
return total
def main():
for _ in range(5):
fast()
slow()
assert hottest_function(main) == "slow"
import cProfile, pstats
def hottest_function(func):
profiler = cProfile.Profile()
profiler.enable()
func()
profiler.disable()
stats = pstats.Stats(profiler).stats
candidates = [(data[2], key[2]) for key, data in stats.items() if key[0] != "~"]
return max(candidates)[1]
С чего правильно начинать оптимизацию медленной программы?
Что даст наибольший эффект?
| сильно ускорит | почти не повлияет | |
|---|---|---|
Заменить поиск x in list в цикле на x in set |
||
| Переименовать переменные покороче | ||
| Убрать N+1 запросов к БД (один запрос с JOIN вместо запроса на каждую строку) | ||
Заменить range(len(x)) на enumerate в цикле на 10 элементов |
||
| Посчитать сумму массива 10⁷ чисел через numpy вместо цикла Python | ||
| Отправлять 1000 HTTP-запросов конкурентно вместо последовательно |
Чем py-spy удобен для профилирования в продакшене?
Войдите, чтобы отмечать пройденные уроки.