Навыки: оптимизация производительности
Когда навык обрабатывает большие файлы или вызывается часто, проблемы с производительностью напрямую влияют на пользовательский опыт.
В этой статье представлены распространённые методы повышения скорости обработки на уровне скрипта.
Распространённые источники проблем с производительностью
| Source | Типичные симптомы | Направление оптимизации |
|---|---|---|
| Загрузка больших файлов целиком | Чтение CSV размером 100MB занимает несколько секунд | Чтение по частям (chunk) |
| Многократное выполнение одного и того же вычисления | Пересчёт статистики при каждом вызове | Кэширование результатов |
| Однопоточная последовательная обработка | Обработка 1000 файлов занимает много времени | Параллельная обработка |
| Частый дисковый ввод-вывод | Запись файлов построчно в цикле | Пакетная запись |
| Загрузка ненужных зависимостей | import занимает больше 2 секунд | Ленивый import |
Чтение больших файлов по частям
Параметрchunksizeв pandas позволяет разбивать большие файлы на несколько пакетов и обрабатывать их по частям, избегая чрезмерного использования памяти за один раз.
Пример
# Путь к файлу: scripts/chunked_reader.py
import pandas as pd
import json
import sys
def process_large_csv(file_path: str, chunk_size: int = 10000) -> dict:
"""
Чтение и обработка больших CSV-файлов по частям
Параметры:
file_path: путь к CSV-файлу
chunk_size: количество строк на блок, по умолчанию 10000
Возвращает:
Объединённые статистические результаты
"""
total_rows = 0
total_sum = 0.0
chunk_count = 0
# Чтение по частям, не загружая весь файл в память сразу
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
chunk_count += 1
total_rows += len(chunk)
# Выполнить статистику по каждому фрагменту, накапливая результаты
if "score" in chunk.columns:
total_sum += chunk["score"].sum()
# Отображать прогресс в реальном времени
print(f" Обработан пакет {chunk_count}, накоплено {total_rows} строк", flush=True)
avg = total_sum / total_rows if total_rows > 0 else 0
return {
"status": "success",
"total_rows": total_rows,
"chunks": chunk_count,
"score_avg": round(avg, 2)
}
if __name__ == "__main__":
file_path = sys.argv[1] if len(sys.argv) > 1 else ""
result = process_large_csv(file_path)
print(json.dumps(result, ensure_ascii=False, indent=2))
import pandas as pd
import json
import sys
def process_large_csv(file_path: str, chunk_size: int = 10000) -> dict:
"""
Чтение и обработка больших CSV-файлов по частям
Параметры:
file_path: путь к CSV-файлу
chunk_size: количество строк на блок, по умолчанию 10000
Возвращает:
Объединённые статистические результаты
"""
total_rows = 0
total_sum = 0.0
chunk_count = 0
# Чтение по частям, не загружая весь файл в память сразу
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
chunk_count += 1
total_rows += len(chunk)
# Выполнить статистику по каждому фрагменту, накапливая результаты
if "score" in chunk.columns:
total_sum += chunk["score"].sum()
# Отображать прогресс в реальном времени
print(f" Обработан пакет {chunk_count}, накоплено {total_rows} строк", flush=True)
avg = total_sum / total_rows if total_rows > 0 else 0
return {
"status": "success",
"total_rows": total_rows,
"chunks": chunk_count,
"score_avg": round(avg, 2)
}
if __name__ == "__main__":
file_path = sys.argv[1] if len(sys.argv) > 1 else ""
result = process_large_csv(file_path)
print(json.dumps(result, ensure_ascii=False, indent=2))
已处理第 1 批，累计 10000 行
已处理第 2 批，累计 20000 行
已处理第 3 批，累计 28456 行
{
"status": "success",
"total_rows": 28456,
"chunks": 3,
"score_avg": 82.37
}
Кэширование результатов
Для операций, где одинаковые входные данные дают одинаковый результат, кэшируйте результат в локальный файл, чтобы избежать повторных вычислений.
Пример
# Путь к файлу: scripts/file_cache.py
import hashlib
import json
import os
import time
CACHE_DIR = "/home/claude/.skill_cache"
os.makedirs(CACHE_DIR, exist_ok=True)
def _file_fingerprint(file_path: str) -> str:
"""
Создать отпечаток на основе пути к файлу, размера и времени модификации
Гораздо быстрее, чем MD5 содержимого файла, подходит для больших файлов
"""
stat = os.stat(file_path)
raw = f"{file_path}|{stat.st_size}|{stat.st_mtime}"
return hashlib.md5(raw.encode()).hexdigest()
def get_cached(file_path: str, operation: str):
"""Получить кэш, вернуть None, если он не существует"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
if os.path.exists(cache_file):
with open(cache_file) as f:
return json.load(f)
return None
def set_cached(file_path: str, operation: str, result: dict):
"""Записать кэш"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
with open(cache_file, "w") as f:
json.dump(result, f, ensure_ascii=False)
# Пример использования
def get_stats(file_path: str) -> dict:
# Сначала проверить кэш
cached = get_cached(file_path, "stats")
if cached:
print("Попадание в кэш, пропускаем повторное вычисление")
return cached
# Промах кэша, выполняем вычисление
import pandas as pd
t0 = time.time()
df = pd.read_csv(file_path)
result = {
"rows": len(df),
"cols": len(df.columns),
"elapsed": round(time.time() - t0, 3)
}
# Записать кэш для следующего использования
set_cached(file_path, "stats", result)
return result
import hashlib
import json
import os
import time
CACHE_DIR = "/home/claude/.skill_cache"
os.makedirs(CACHE_DIR, exist_ok=True)
def _file_fingerprint(file_path: str) -> str:
"""
Создать отпечаток на основе пути к файлу, размера и времени модификации
Гораздо быстрее, чем MD5 содержимого файла, подходит для больших файлов
"""
stat = os.stat(file_path)
raw = f"{file_path}|{stat.st_size}|{stat.st_mtime}"
return hashlib.md5(raw.encode()).hexdigest()
def get_cached(file_path: str, operation: str):
"""Получить кэш, вернуть None, если он не существует"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
if os.path.exists(cache_file):
with open(cache_file) as f:
return json.load(f)
return None
def set_cached(file_path: str, operation: str, result: dict):
"""Записать кэш"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
with open(cache_file, "w") as f:
json.dump(result, f, ensure_ascii=False)
# Пример использования
def get_stats(file_path: str) -> dict:
# Сначала проверить кэш
cached = get_cached(file_path, "stats")
if cached:
print("Попадание в кэш, пропускаем повторное вычисление")
return cached
# Промах кэша, выполняем вычисление
import pandas as pd
t0 = time.time()
df = pd.read_csv(file_path)
result = {
"rows": len(df),
"cols": len(df.columns),
"elapsed": round(time.time() - t0, 3)
}
# Записать кэш для следующего использования
set_cached(file_path, "stats", result)
return result
Параллельная обработка нескольких файлов
Когда нужно обработать несколько файлов, используйтеconcurrent.futuresдля параллельного выполнения, что может ускорить работу в несколько раз.
Пример
# Путь к файлу: scripts/parallel_process.py
import concurrent.futures
import os
import json
import sys
def process_single_file(file_path: str) -> dict:
"""Обрабатывает один файл (будет вызываться параллельно)"""
try:
size = os.path.getsize(file_path)
# Имитация фактической логики обработки
return {"file": os.path.basename(file_path),
"size_kb": size // 1024, "status": "ok"}
except Exception as e:
return {"file": file_path, "status": "error", "message": str(e)}
def process_files_parallel(file_paths: list, max_workers: int = 4) -> list:
"""
Параллельная обработка нескольких файлов
Параметры:
file_paths: список путей к файлам
max_workers: максимальное количество одновременных задач, по умолчанию 4 (чтобы избежать конкуренции слишком большого числа процессов за ресурсы)
"""
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Отправить все задачи
future_map = {
executor.submit(process_single_file, fp): fp
for fp in file_paths
}
# Собрать результаты в порядке завершения
for future in concurrent.futures.as_completed(future_map):
result = future.result()
results.append(result)
print(f" Завершено: {result['file']}")
return results
if __name__ == "__main__":
# Имитация: обработать все CSV-файлы в каталоге загрузки
upload_dir = "/mnt/user-data/uploads"
csv_files = [
os.path.join(upload_dir, f)
for f in os.listdir(upload_dir)
if f.endswith(".csv")
]
if not csv_files:
print("CSV-файлы не найдены")
sys.exit(0)
print(f"Запуск параллельной обработки {len(csv_files)} файлов...")
results = process_files_parallel(csv_files)
print(json.dumps(results, ensure_ascii=False, indent=2))
import concurrent.futures
import os
import json
import sys
def process_single_file(file_path: str) -> dict:
"""Обрабатывает один файл (будет вызываться параллельно)"""
try:
size = os.path.getsize(file_path)
# Имитация фактической логики обработки
return {"file": os.path.basename(file_path),
"size_kb": size // 1024, "status": "ok"}
except Exception as e:
return {"file": file_path, "status": "error", "message": str(e)}
def process_files_parallel(file_paths: list, max_workers: int = 4) -> list:
"""
Параллельная обработка нескольких файлов
Параметры:
file_paths: список путей к файлам
max_workers: максимальное количество одновременных задач, по умолчанию 4 (чтобы избежать конкуренции слишком большого числа процессов за ресурсы)
"""
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Отправить все задачи
future_map = {
executor.submit(process_single_file, fp): fp
for fp in file_paths
}
# Собрать результаты в порядке завершения
for future in concurrent.futures.as_completed(future_map):
result = future.result()
results.append(result)
print(f" Завершено: {result['file']}")
return results
if __name__ == "__main__":
# Имитация: обработать все CSV-файлы в каталоге загрузки
upload_dir = "/mnt/user-data/uploads"
csv_files = [
os.path.join(upload_dir, f)
for f in os.listdir(upload_dir)
if f.endswith(".csv")
]
if not csv_files:
print("CSV-файлы не найдены")
sys.exit(0)
print(f"Запуск параллельной обработки {len(csv_files)} файлов...")
results = process_files_parallel(csv_files)
print(json.dumps(results, ensure_ascii=False, indent=2))
开始并行处理 3 个文件...
完成：example_jan.csv
完成：example_mar.csv
完成：example_feb.csv
[{"file": "example_jan.csv", "size_kb": 128, "status": "ok"}, ...]
Чек-лист по оптимизации производительности
| Пункт проверки | До оптимизации | После оптимизации |
|---|---|---|
| Чтение больших файлов | pd.read_csv(file) | pd.read_csv(file, chunksize=10000) |
| Многократное вычисление статистики по одному и тому же файлу | Пересчёт каждый раз | Использование отпечатка файла для кэширования результатов |
| Обработка нескольких файлов | цикл for, последовательно | ThreadPoolExecutor, параллельно |
| Запись файлов построчно | for row: f.write(row) | Сбор пакетами и запись за один раз |
| Ненужные импорты | Все импорты в начале скрипта | Импорт при необходимости (ленивый импорт) |
Другие расширенияИзмеряйте перед оптимизацией; не оптимизируйте по интуиции. Используйте
time.perf_counter()или@timeitдекоратор, чтобы выявить реальное узкое место, затем оптимизируйте соответственно, избегая ненужного усложнения кода из-за преждевременной оптимизации.