본문으로 건너뛰기
Advertisement

메모리 최적화

__slots__, 제너레이터, 약한 참조로 메모리 사용량을 줄입니다.


__slots__ — 인스턴스 딕셔너리 제거

import sys
from typing import ClassVar


# ── 일반 클래스 vs __slots__ ─────────────────────────────
class PointNormal:
def __init__(self, x: float, y: float, z: float):
self.x = x
self.y = y
self.z = z


class PointSlots:
__slots__ = ("x", "y", "z") # 허용 속성 목록

def __init__(self, x: float, y: float, z: float):
self.x = x
self.y = y
self.z = z


p1 = PointNormal(1.0, 2.0, 3.0)
p2 = PointSlots(1.0, 2.0, 3.0)

print(sys.getsizeof(p1)) # ~48 bytes (+ __dict__ ~232 bytes)
print(sys.getsizeof(p2)) # ~64 bytes (딕셔너리 없음)
print(hasattr(p1, "__dict__")) # True
print(hasattr(p2, "__dict__")) # False

# 100만 개 객체 비교
normal_objects = [PointNormal(float(i), float(i), float(i)) for i in range(1_000_000)]
slots_objects = [PointSlots(float(i), float(i), float(i)) for i in range(1_000_000)]

# __slots__ 제약: 동적 속성 추가 불가
# p2.w = 4.0 # AttributeError!

dataclass + __slots__ (Python 3.10+)

from dataclasses import dataclass


@dataclass(slots=True) # Python 3.10+: 자동으로 __slots__ 적용
class Vector3D:
x: float
y: float
z: float

def magnitude(self) -> float:
return (self.x ** 2 + self.y ** 2 + self.z ** 2) ** 0.5

def __add__(self, other: "Vector3D") -> "Vector3D":
return Vector3D(self.x + other.x, self.y + other.y, self.z + other.z)


v1 = Vector3D(1.0, 2.0, 3.0)
v2 = Vector3D(4.0, 5.0, 6.0)
print(v1 + v2) # Vector3D(x=5.0, y=7.0, z=9.0)
print(v1.magnitude()) # 3.74...


# frozen=True: 불변 객체 (hashable, 더 안전)
@dataclass(frozen=True, slots=True)
class ImmutablePoint:
x: float
y: float

약한 참조 (weakref) — 순환 참조 방지

import weakref
import gc


class Node:
def __init__(self, value: int):
self.value = value
self.children: list["Node"] = []
self._parent: weakref.ref["Node"] | None = None

@property
def parent(self) -> "Node | None":
return self._parent() if self._parent else None

@parent.setter
def parent(self, node: "Node") -> None:
self._parent = weakref.ref(node) # 약한 참조 → 순환 참조 없음


root = Node(0)
child = Node(1)
child.parent = root
root.children.append(child)

# root가 삭제되면 child.parent()는 None 반환
del root
print(child.parent) # None (약한 참조 → GC 수거됨)

# WeakValueDictionary — 값이 다른 곳에서 참조되지 않으면 자동 제거
cache: weakref.WeakValueDictionary[str, Node] = weakref.WeakValueDictionary()
node = Node(42)
cache["node42"] = node
print("node42" in cache) # True
del node
gc.collect()
print("node42" in cache) # False (자동 제거)

array 모듈 — 타입화 배열

import array
import sys

# list — 파이썬 객체 (오버헤드 큼)
py_list = list(range(1_000_000))
print(f"list: {sys.getsizeof(py_list) / 1024 / 1024:.1f} MB") # ~8 MB

# array — C 배열 (raw bytes)
arr = array.array("d", range(1_000_000)) # "d" = double (8 bytes)
print(f"array: {sys.getsizeof(arr) / 1024 / 1024:.1f} MB") # ~8 MB (헤더 절약)

# 타입 코드
# "b" = signed char (1 byte), "i" = int (4 bytes), "d" = double (8 bytes)
# "f" = float (4 bytes), "l" = long (8 bytes)

int_arr = array.array("i", [1, 2, 3, 4, 5])
int_arr.append(6)
int_arr.extend([7, 8, 9])
print(sum(int_arr))

NumPy vs list 메모리 비교

import numpy as np
import sys

# Python list of floats
py_list = [float(i) for i in range(1_000_000)]

# NumPy array
np_arr = np.arange(1_000_000, dtype=np.float64)

py_mem = sum(sys.getsizeof(x) for x in py_list) + sys.getsizeof(py_list)
np_mem = np_arr.nbytes

print(f"Python list: {py_mem / 1024 / 1024:.1f} MB") # ~28 MB
print(f"NumPy array: {np_mem / 1024 / 1024:.1f} MB") # ~8 MB

# 메모리 효율적인 NumPy dtype 선택
data_int64 = np.zeros(1_000_000, dtype=np.int64) # 8MB
data_int32 = np.zeros(1_000_000, dtype=np.int32) # 4MB
data_int16 = np.zeros(1_000_000, dtype=np.int16) # 2MB (범위: -32768~32767)
data_int8 = np.zeros(1_000_000, dtype=np.int8) # 1MB (범위: -128~127)

# dtype 다운캐스팅
import pandas as pd

df = pd.DataFrame({"value": range(100_000)})
print(df.dtypes) # int64
df["value"] = pd.to_numeric(df["value"], downcast="integer")
print(df.dtypes) # int8 or int16 (범위 맞는 최소 타입)

제너레이터 파이프라인

from pathlib import Path


# 전체 파일을 메모리에 올리지 않고 스트리밍 처리
def read_lines(filepath: str):
"""파일 라인 스트리밍"""
with open(filepath, encoding="utf-8") as f:
yield from f


def parse_csv_row(lines):
"""CSV 파싱"""
for line in lines:
yield line.strip().split(",")


def filter_valid(rows):
"""유효한 행만 통과"""
for row in rows:
if len(row) >= 3 and row[0].strip():
yield row


def transform(rows):
"""데이터 변환"""
for row in rows:
yield {
"id": int(row[0]),
"name": row[1].strip(),
"value": float(row[2]),
}


# 파이프라인 조합 — 한 번에 한 줄씩 처리
def process_large_file(filepath: str):
pipeline = transform(
filter_valid(
parse_csv_row(
read_lines(filepath)
)
)
)
for record in pipeline:
yield record # 또는 DB 저장, API 전송 등

정리

기법메모리 절약적용 조건
__slots__20~50% (객체당)다수의 동일 클래스 인스턴스
weakref순환 참조 제거부모-자식, 캐시
array 모듈60~80% vs list숫자 데이터 단순 배열
NumPy dtype 최적화50~87%데이터 범위가 좁을 때
제너레이터 파이프라인99%+ (스트리밍)대용량 파일/데이터 처리
Advertisement