파이썬 코드 리팩토링 마스터 가이드 - Part 2: 성능 최적화 핵심 가이드 💫

들어가며 🚀

파이썬 코드의 성능을 개선하는 구체적인 기법들을 하나씩 살펴보겠습니다. 특히 실제 현업에서 자주 마주치는 성능 이슈들을 중심으로 설명하겠습니다.

1. 데이터 처리 최적화 (Data Processing Optimization) 🔄

1.1 대용량 데이터 처리 최적화

# 개선 전 - 메모리 문제 발생
def process_large_csv(filename):
    with open(filename) as f:
        data = f.readlines()  # 🚫 전체 파일을 메모리에 로드

    results = []
    for line in data:
        if float(line.split(',')[1]) > 100:
            results.append(line)
    return results

# 개선 후 - 메모리 효율적 처리
def process_large_csv(filename):
    """대용량 CSV 파일을 메모리 효율적으로 처리합니다."""
    def process_chunk():
        with open(filename) as f:
            for line in f:  # ✅ 한 줄씩 처리
                value = float(line.split(',')[1])
                if value > 100:
                    yield line.strip()

    return process_chunk()

# 실제 사용 예시와 메모리 사용량 비교
import memory_profiler

@memory_profiler.profile
def main():
    data = list(process_large_csv('large_data.csv'))
    print(f"처리된 데이터 수: {len(data)}")

1.2 데이터 집계 최적화

from collections import Counter, defaultdict
from typing import Dict, List
import pandas as pd

class DataAggregator:
    """데이터 집계를 최적화하는 클래스입니다."""

    def __init__(self, data: List[Dict]):
        self.data = data
        self.cache = {}  # 계산 결과 캐싱

    def aggregate_by_category(self, category_field: str) -> Dict:
        """카테고리별 집계를 수행합니다."""
        cache_key = f"category_{category_field}"

        if cache_key in self.cache:
            return self.cache[cache_key]

        # Counter 사용으로 집계 최적화
        result = Counter(item[category_field] for item in self.data)
        self.cache[cache_key] = dict(result)

        return self.cache[cache_key]

    def calculate_statistics(self, value_field: str) -> Dict:
        """필드별 통계를 계산합니다."""
        # Pandas 사용으로 통계 계산 최적화
        df = pd.DataFrame(self.data)
        stats = {
            'mean': df[value_field].mean(),
            'median': df[value_field].median(),
            'std': df[value_field].std(),
            'min': df[value_field].min(),
            'max': df[value_field].max()
        }
        return stats

# 성능 측정
import time

def measure_performance(func):
    """함수의 실행 시간을 측정하는 데코레이터입니다."""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} 실행 시간: {end_time - start_time:.4f}초")
        return result
    return wrapper

2. 병렬 처리 최적화 (Parallel Processing) ⚡

2.1 멀티프로세싱을 활용한 병렬 처리

from multiprocessing import Pool
import time
from typing import List

class ParallelDataProcessor:
    """데이터 병렬 처리를 위한 클래스입니다."""

    def __init__(self, process_count: int = 4):
        self.process_count = process_count

    @staticmethod
    def heavy_computation(data: int) -> int:
        """시간이 오래 걸리는 계산을 시뮬레이션합니다."""
        time.sleep(0.1)  # 복잡한 계산 시뮬레이션
        return data * data

    def process_sequential(self, data_list: List[int]) -> List[int]:
        """순차적 처리를 수행합니다."""
        return [self.heavy_computation(x) for x in data_list]

    def process_parallel(self, data_list: List[int]) -> List[int]:
        """병렬 처리를 수행합니다."""
        with Pool(processes=self.process_count) as pool:
            return pool.map(self.heavy_computation, data_list)

# 성능 비교
@measure_performance
def compare_processing_methods():
    processor = ParallelDataProcessor()
    data = list(range(100))

    # 순차 처리
    sequential_start = time.time()
    sequential_result = processor.process_sequential(data)
    sequential_time = time.time() - sequential_start

    # 병렬 처리
    parallel_start = time.time()
    parallel_result = processor.process_parallel(data)
    parallel_time = time.time() - parallel_start

    print(f"순차 처리 시간: {sequential_time:.2f}초")
    print(f"병렬 처리 시간: {parallel_time:.2f}초")
    print(f"성능 향상: {sequential_time/parallel_time:.1f}배")

2.2 비동기 I/O 최적화

import asyncio
import aiohttp
from typing import List, Dict
import logging

class AsyncDataFetcher:
    """비동기 데이터 조회를 위한 클래스입니다."""

    def __init__(self, base_url: str, max_concurrent: int = 10):
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.session = None
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.logger = logging.getLogger(__name__)

    async def __aenter__(self):
        """비동기 컨텍스트 매니저 진입점."""
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        """비동기 컨텍스트 매니저 종료점."""
        await self.session.close()

    async def fetch_with_retry(self, url: str, retries: int = 3) -> Dict:
        """재시도 로직이 포함된 데이터 조회."""
        for attempt in range(retries):
            try:
                async with self.semaphore:
                    async with self.session.get(url) as response:
                        if response.status == 200:
                            return await response.json()
                        else:
                            self.logger.warning(
                                f"Failed to fetch {url}, "
                                f"status: {response.status}"
                            )
            except Exception as e:
                if attempt == retries - 1:
                    self.logger.error(f"Failed to fetch {url}: {str(e)}")
                    raise
                await asyncio.sleep(2 ** attempt)  # 지수 백오프

    async def fetch_all(self, endpoints: List[str]) -> List[Dict]:
        """여러 엔드포인트의 데이터를 동시에 조회."""
        urls = [f"{self.base_url}/{endpoint}" for endpoint in endpoints]
        tasks = [self.fetch_with_retry(url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 사용 예시
async def main():
    async with AsyncDataFetcher("http://api.example.com") as fetcher:
        endpoints = [f"data/{i}" for i in range(100)]
        results = await fetcher.fetch_all(endpoints)

    success = sum(1 for r in results if not isinstance(r, Exception))
    print(f"성공: {success}/{len(results)}")

3. 메모리 최적화 (Memory Optimization) 💾

3.1 메모리 사용량 모니터링

import psutil
import os
from typing import Callable
from functools import wraps

def memory_usage_decorator(func: Callable):
    """함수의 메모리 사용량을 모니터링하는 데코레이터입니다."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        process = psutil.Process(os.getpid())
        mem_before = process.memory_info().rss / 1024 / 1024  # MB

        result = func(*args, **kwargs)

        mem_after = process.memory_info().rss / 1024 / 1024
        print(f"메모리 사용량 변화: {mem_after - mem_before:.2f} MB")
        return result
    return wrapper

@memory_usage_decorator
def process_data(data: List[Dict]) -> Dict:
    """데이터 처리 함수 예시입니다."""
    result = defaultdict(list)
    for item in data:
        result[item['category']].append(item['value'])
    return dict(result)

3.2 캐시 최적화

from functools import lru_cache
import time
from typing import Dict, Any

class CacheOptimizer:
    """캐시 최적화를 위한 클래스입니다."""

    def __init__(self, cache_size: int = 128):
        self.cache_size = cache_size
        self.cache_stats = {'hits': 0, 'misses': 0}

    @lru_cache(maxsize=128)
    def expensive_operation(self, key: str) -> Any:
        """비용이 많이 드는 작업을 캐싱합니다."""
        time.sleep(1)  # 복잡한 연산 시뮬레이션
        return f"결과_{key}"

    def get_cache_stats(self) -> Dict:
        """캐시 통계를 반환합니다."""
        info = self.expensive_operation.cache_info()
        return {
            'hits': info.hits,
            'misses': info.misses,
            'cache_size': info.maxsize,
            'current_size': info.currsize
        }

    def clear_cache(self):
        """캐시를 초기화합니다."""
        self.expensive_operation.cache_clear()

정리 🎁

Part 2에서는 파이썬 코드의 성능을 최적화하는 핵심 기법들을 살펴보았습니다:

1. 데이터 처리 최적화

   • 대용량 데이터의 메모리 효율적 처리
   • 효율적인 데이터 집계 방법

2. 병렬 처리 최적화

   • 멀티프로세싱을 활용한 CPU 바운드 작업 최적화
   • 비동기 I/O를 활용한 네트워크 작업 최적화

3. 메모리 최적화

   • 메모리 사용량 모니터링
   • 캐시를 활용한 성능 개선

다음 Part 3에서는 코드의 테스트 용이성과 유지보수성을 높이는 리팩토링 기법들을 다루겠습니다.

---

References:

1. High Performance Python (Micha Gorelick & Ian Ozsvald) - Chapter 2, 4, 7
2. Python Cookbook (David Beazley & Brian K. Jones) - Chapter 14: Testing, Debugging, and Exceptions
3. asyncio Documentation (https://docs.python.org/3/library/asyncio.html)
4. Python Performance Tips (https://wiki.python.org/moin/PythonSpeed/PerformanceTips)
5. Memory Profiler Documentation (https://pypi.org/project/memory-profiler/)