pickle과 이진 파일

pickle은 Python 객체를 바이트 스트림으로 직렬화(serialize)하고, 다시 복원(deserialize)하는 표준 라이브러리입니다. Python 객체를 파일에 저장하거나 네트워크로 전송할 때 사용합니다.

---

pickle 기본: dumps/loads, dump/load

import pickle


# --- 메모리 내 직렬화/역직렬화 ---

# 간단한 객체들
data = {
    "name": "김철수",
    "age": 30,
    "scores": [95, 87, 92],
    "metadata": {"active": True, "role": "admin"},
}

# 객체 → 바이트 스트림
serialized = pickle.dumps(data)
print(type(serialized))        # <class 'bytes'>
print(len(serialized))         # 바이트 수

# 바이트 스트림 → 객체
restored = pickle.loads(serialized)
print(restored == data)        # True
print(restored is data)        # False (새 객체 생성)

# --- 파일에 저장/로드 ---

# 저장
with open("data.pkl", "wb") as f:   # 반드시 바이너리 모드 "wb"
    pickle.dump(data, f)

# 로드
with open("data.pkl", "rb") as f:   # 바이너리 읽기 "rb"
    loaded = pickle.load(f)

print(loaded)


# --- 여러 객체를 하나의 파일에 저장 ---

records = [
    {"id": 1, "value": "first"},
    {"id": 2, "value": "second"},
    {"id": 3, "value": "third"},
]

with open("records.pkl", "wb") as f:
    for record in records:
        pickle.dump(record, f)  # 연속으로 dump 가능

# 연속으로 load
loaded_records = []
with open("records.pkl", "rb") as f:
    while True:
        try:
            loaded_records.append(pickle.load(f))
        except EOFError:
            break  # 파일 끝에 도달

print(loaded_records)

---

pickle 프로토콜 버전

import pickle
import sys


print(f"기본 프로토콜: {pickle.DEFAULT_PROTOCOL}")   # 보통 5
print(f"최고 프로토콜: {pickle.HIGHEST_PROTOCOL}")   # Python 버전에 따라 다름

data = list(range(10_000))

# 프로토콜별 크기 비교
for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
    serialized = pickle.dumps(data, protocol=protocol)
    print(f"프로토콜 {protocol}: {len(serialized):,} bytes")

# 특정 프로토콜로 저장
with open("data_v5.pkl", "wb") as f:
    pickle.dump(data, f, protocol=5)

# pickle.HIGHEST_PROTOCOL 사용 권장 (최신 Python 전용)
# 호환성이 필요하면 낮은 프로토콜 사용
with open("data_compat.pkl", "wb") as f:
    pickle.dump(data, f, protocol=2)  # Python 2.3+와 호환

---

사용자 정의 클래스 직렬화

import pickle
from datetime import datetime


class Employee:
    """pickle로 직렬화 가능한 클래스"""

    def __init__(self, emp_id: str, name: str, salary: float, hire_date: datetime):
        self.emp_id = emp_id
        self.name = name
        self.salary = salary
        self.hire_date = hire_date
        self._cache: dict = {}  # 캐시 — 직렬화 시 제외하고 싶을 수 있음

    def __repr__(self) -> str:
        return f"Employee(id={self.emp_id!r}, name={self.name!r})"


# 기본 직렬화: __dict__의 모든 속성 포함
emp = Employee("E001", "김철수", 5_000_000, datetime(2020, 3, 15))
serialized = pickle.dumps(emp)
restored = pickle.loads(serialized)
print(restored)
print(restored.salary)  # 5000000.0


# __getstate__ / __setstate__로 직렬화 제어
class OptimizedEmployee:
    def __init__(self, emp_id: str, name: str, salary: float):
        self.emp_id = emp_id
        self.name = name
        self.salary = salary
        self._cache: dict = {}          # 캐시 (직렬화 불필요)
        self._connection = None         # 연결 객체 (직렬화 불가)

    def __getstate__(self) -> dict:
        """직렬화할 상태를 반환 — _cache와 _connection 제외"""
        state = self.__dict__.copy()
        del state["_cache"]
        del state["_connection"]
        return state

    def __setstate__(self, state: dict) -> None:
        """역직렬화 시 상태 복원 — 제외된 속성 재초기화"""
        self.__dict__.update(state)
        self._cache = {}        # 빈 캐시로 재초기화
        self._connection = None # 연결은 별도로 재설정 필요

    def __repr__(self) -> str:
        return f"OptimizedEmployee({self.emp_id!r}, {self.name!r})"


oe = OptimizedEmployee("E002", "이영희", 6_000_000)
oe._cache = {"key": "value"}  # 캐시 데이터

serialized = pickle.dumps(oe)
restored = pickle.loads(serialized)

print(restored)
print(restored._cache)       # {} (재초기화됨)
print(restored._connection)  # None (재초기화됨)

---

직렬화 가능/불가 객체 구분

import pickle


# ✅ 직렬화 가능한 객체들
serializable_objects = [
    None,
    True, False,
    42, 3.14, 1 + 2j,
    "문자열",
    b"바이트",
    [1, 2, 3],
    (1, "a"),
    {"key": "value"},
    {1, 2, 3},
    range(10),
]

for obj in serializable_objects:
    try:
        data = pickle.dumps(obj)
        restored = pickle.loads(data)
        print(f"✅ {type(obj).__name__}: {obj!r} → 직렬화 성공")
    except Exception as e:
        print(f"❌ {type(obj).__name__}: {e}")


# ❌ 직렬화 불가능한 객체들
import threading
import io


def my_function():
    pass  # 람다와 달리 일반 함수는 직렬화 가능


non_serializable = [
    lambda x: x * 2,          # 람다 함수는 직렬화 불가
    threading.Lock(),          # 스레드 락
    io.StringIO("test"),       # 일부 IO 객체
]

for obj in non_serializable:
    try:
        pickle.dumps(obj)
        print(f"✅ {type(obj).__name__} 직렬화 성공")
    except (pickle.PicklingError, AttributeError, TypeError) as e:
        print(f"❌ {type(obj).__name__}: {e}")


# 함수와 클래스는 모듈 레벨에 정의된 경우 직렬화 가능
import pickle

def add(a, b):
    return a + b

serialized_func = pickle.dumps(add)  # 모듈 레벨 함수는 OK
restored_func = pickle.loads(serialized_func)
print(restored_func(3, 4))  # 7

---

보안 주의사항

import pickle


# ⚠️ 경고: pickle.loads()는 절대 신뢰할 수 없는 데이터에 사용하면 안 됩니다!
# pickle 역직렬화 시 임의 코드가 실행될 수 있습니다.

# 악의적인 pickle 데이터 예시 (실행하지 마세요!)
# 실제로는 아래와 같은 코드가 역직렬화 시 실행될 수 있음:
#
# class Exploit:
#     def __reduce__(self):
#         return (os.system, ("rm -rf /",))  # 시스템 명령 실행!
#
# malicious = pickle.dumps(Exploit())
# pickle.loads(malicious)  # rm -rf / 실행됨!


# ✅ 안전한 사용 지침:
# 1. 자신의 코드에서 생성한 pickle 파일만 로드
# 2. 외부에서 받은 pickle 데이터는 절대 로드 금지
# 3. 네트워크 통신에 pickle 사용 금지
# 4. 사용자 입력으로 생성된 pickle 데이터 사용 금지

# ✅ 대안 — JSON은 코드 실행 없이 안전하게 역직렬화
import json

safe_data = {"name": "김철수", "age": 30, "scores": [95, 87]}
json_str = json.dumps(safe_data, ensure_ascii=False)
restored = json.loads(json_str)  # 안전: JSON은 코드를 실행하지 않음
print(restored)

# hmac을 사용한 pickle 서명으로 무결성 검증
import hmac
import hashlib


def secure_dumps(obj, secret_key: bytes) -> tuple[bytes, bytes]:
    """객체를 직렬화하고 HMAC 서명 추가"""
    data = pickle.dumps(obj)
    signature = hmac.new(secret_key, data, hashlib.sha256).digest()
    return data, signature


def secure_loads(data: bytes, signature: bytes, secret_key: bytes):
    """HMAC 서명 검증 후 역직렬화"""
    expected = hmac.new(secret_key, data, hashlib.sha256).digest()
    if not hmac.compare_digest(signature, expected):
        raise ValueError("데이터가 변조되었습니다!")
    return pickle.loads(data)


secret = b"my-secret-key-32-bytes-long!!!!!"
original = {"user": "admin", "role": "superuser"}

data, sig = secure_dumps(original, secret)
restored = secure_loads(data, sig, secret)
print(f"서명 검증 통과: {restored}")

# 변조 시도
tampered_data = data[:-1] + bytes([data[-1] ^ 0xFF])
try:
    secure_loads(tampered_data, sig, secret)
except ValueError as e:
    print(f"변조 감지: {e}")

---

shelve 모듈로 영속적 저장

shelve는 pickle 기반의 영속적 딕셔너리입니다. 키-값 형태로 Python 객체를 파일에 저장하고 조회할 수 있습니다.

import shelve
from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class UserProfile:
    user_id: str
    username: str
    email: str
    created_at: datetime = field(default_factory=datetime.now)
    preferences: dict = field(default_factory=dict)

    def __repr__(self) -> str:
        return f"UserProfile({self.username!r})"


# shelve 사용 — 딕셔너리처럼 사용
with shelve.open("users_db") as db:
    # 저장
    user1 = UserProfile("U001", "alice", "alice@example.com")
    user2 = UserProfile("U002", "bob", "bob@example.com",
                        preferences={"theme": "dark", "lang": "ko"})

    db["U001"] = user1
    db["U002"] = user2
    db["U001"].preferences["theme"] = "light"  # 주의: 수정 후 재할당 필요

    # 재할당 없이 수정 시 변경이 저장되지 않을 수 있음
    # 안전한 방법:
    u = db["U001"]
    u.preferences["notifications"] = True
    db["U001"] = u  # 명시적 재할당

    print(f"저장된 사용자 수: {len(db)}")
    print(f"키 목록: {list(db.keys())}")


# 나중에 파일에서 다시 읽기
with shelve.open("users_db") as db:
    for user_id, user in db.items():
        print(f"  {user_id}: {user}")

    # 특정 키 조회
    if "U001" in db:
        alice = db["U001"]
        print(f"Alice 선호 설정: {alice.preferences}")

    # 삭제
    if "U002" in db:
        del db["U002"]
        print(f"U002 삭제 후: {list(db.keys())}")


# writeback=True 옵션 — 수정 사항 자동 감지 (메모리 증가 주의)
with shelve.open("users_db", writeback=True) as db:
    if "U001" in db:
        db["U001"].preferences["auto_save"] = True  # writeback=True면 재할당 불필요
        # 단, 모든 항목이 메모리에 캐싱되어 메모리 사용량 증가

---

struct 모듈로 이진 파일 처리

struct 모듈은 C 언어의 구조체처럼 이진 데이터를 팩/언팩합니다. 파일 포맷 분석, 네트워크 프로토콜, 하드웨어 통신 등에 사용됩니다.

import struct


# struct 형식 문자열
# '>' 빅 엔디안, '<' 리틀 엔디안, '=' 네이티브 엔디안
# 'i' 4바이트 정수, 'f' 4바이트 float, 'd' 8바이트 double
# 'c' 1바이트 문자, 's' 문자열, 'B' 1바이트 부호없는 정수

# 팩 (Python → bytes)
fmt = ">ifd"  # 빅 엔디안: int(4) + float(4) + double(8) = 16 bytes
packed = struct.pack(fmt, 42, 3.14, 2.718281828)
print(f"팩 결과: {packed.hex()}")
print(f"크기: {struct.calcsize(fmt)} bytes")

# 언팩 (bytes → Python)
unpacked = struct.unpack(fmt, packed)
print(f"언팩 결과: {unpacked}")  # (42, 3.14..., 2.718...)

# struct를 사용한 이진 파일 쓰기/읽기
RECORD_FORMAT = ">I10sf"  # 부호없는 int(4) + 10바이트 이름 + float(4) = 18 bytes
RECORD_SIZE = struct.calcsize(RECORD_FORMAT)

# 센서 데이터 기록
sensor_data = [
    (1001, b"Sensor_A  ", 23.5),
    (1002, b"Sensor_B  ", 18.2),
    (1003, b"Sensor_C  ", 35.7),
]

with open("sensors.bin", "wb") as f:
    for sensor_id, name, temperature in sensor_data:
        record = struct.pack(RECORD_FORMAT, sensor_id, name, temperature)
        f.write(record)

# 이진 파일 읽기
records_read = []
with open("sensors.bin", "rb") as f:
    while True:
        raw = f.read(RECORD_SIZE)
        if len(raw) < RECORD_SIZE:
            break
        sensor_id, name, temperature = struct.unpack(RECORD_FORMAT, raw)
        records_read.append({
            "id": sensor_id,
            "name": name.rstrip(b"\x00 ").decode("ascii"),
            "temperature": round(temperature, 1),
        })

print("읽은 센서 데이터:")
for record in records_read:
    print(f"  {record}")

# 특정 위치 랜덤 접근 (파일이 고정 크기 레코드이므로 가능)
with open("sensors.bin", "rb") as f:
    # 두 번째 레코드 직접 접근
    f.seek(RECORD_SIZE * 1)
    raw = f.read(RECORD_SIZE)
    sensor_id, name, temp = struct.unpack(RECORD_FORMAT, raw)
    print(f"\n2번째 레코드: ID={sensor_id}, 온도={temp:.1f}°C")

---

대안 라이브러리

joblib — 대용량 배열/ML 모델

# pip install joblib
import joblib
import numpy as np  # pip install numpy


# joblib은 NumPy 배열을 효율적으로 직렬화 (pickle보다 빠름)
data = {
    "weights": np.random.randn(1000, 1000),  # 대용량 NumPy 배열
    "labels": np.array([0, 1, 2, 3]),
    "config": {"learning_rate": 0.001},
}

# 저장
joblib.dump(data, "model_data.joblib")

# 메모리 맵 로드 (파일을 메모리에 완전히 올리지 않고 접근)
loaded = joblib.load("model_data.joblib", mmap_mode="r")
print(f"weights shape: {loaded['weights'].shape}")

# 압축 레벨 지정 (0=없음, 9=최대)
joblib.dump(data, "model_compressed.joblib", compress=3)

msgpack — 빠른 크로스 언어 직렬화

# pip install msgpack
import msgpack


# msgpack은 JSON과 비슷하지만 더 작고 빠름
# 다른 언어(JavaScript, Java 등)와 데이터 교환에 적합
data = {
    "user_id": 12345,
    "name": "김철수",
    "scores": [95, 87, 92, 88],
    "active": True,
}

# 직렬화
packed = msgpack.packb(data, use_bin_type=True)
print(f"msgpack 크기: {len(packed)} bytes")

# JSON과 크기 비교
import json
json_str = json.dumps(data, ensure_ascii=False).encode()
print(f"JSON 크기:    {len(json_str)} bytes")

# 역직렬화
unpacked = msgpack.unpackb(packed, raw=False)
print(f"복원: {unpacked}")

dill — 람다와 클로저도 직렬화

# pip install dill
import dill


# dill은 pickle보다 더 많은 Python 객체를 직렬화 가능
# 람다, 클로저, 중첩 함수 등 지원

multiplier = lambda x: x * 3  # 일반 pickle은 람다 직렬화 불가

def make_adder(n: int):
    def adder(x: int) -> int:
        return x + n
    return adder

add5 = make_adder(5)  # 클로저

# dill로 직렬화
serialized_lambda = dill.dumps(multiplier)
serialized_closure = dill.dumps(add5)

# 역직렬화
restored_lambda = dill.loads(serialized_lambda)
restored_closure = dill.loads(serialized_closure)

print(restored_lambda(7))   # 21
print(restored_closure(10)) # 15

# 멀티프로세싱에서 람다를 사용할 때 유용
# multiprocessing.Pool은 기본적으로 pickle을 사용하는데,
# dill을 사용하면 람다도 다른 프로세스로 전달 가능

---

실전 예제: 체크포인트 시스템

import pickle
import os
import hashlib
from datetime import datetime
from pathlib import Path


class Checkpoint:
    """학습/계산 중간 결과를 저장하는 체크포인트 시스템"""

    def __init__(self, checkpoint_dir: str = "./checkpoints"):
        self.checkpoint_dir = Path(checkpoint_dir)
        self.checkpoint_dir.mkdir(parents=True, exist_ok=True)

    def save(self, name: str, state: dict, metadata: dict | None = None) -> Path:
        """상태를 체크포인트로 저장"""
        checkpoint = {
            "name": name,
            "state": state,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
            "version": 1,
        }

        filepath = self.checkpoint_dir / f"{name}.pkl"
        with open(filepath, "wb") as f:
            pickle.dump(checkpoint, f, protocol=pickle.HIGHEST_PROTOCOL)

        # 체크섬 저장
        checksum = self._compute_checksum(filepath)
        checksum_file = self.checkpoint_dir / f"{name}.sha256"
        checksum_file.write_text(checksum)

        print(f"체크포인트 저장: {filepath} (체크섬: {checksum[:8]}...)")
        return filepath

    def load(self, name: str) -> dict:
        """체크포인트를 로드하고 무결성 검증"""
        filepath = self.checkpoint_dir / f"{name}.pkl"
        checksum_file = self.checkpoint_dir / f"{name}.sha256"

        if not filepath.exists():
            raise FileNotFoundError(f"체크포인트 없음: {name}")

        # 무결성 검증
        if checksum_file.exists():
            expected = checksum_file.read_text().strip()
            actual = self._compute_checksum(filepath)
            if actual != expected:
                raise ValueError(f"체크포인트 손상됨: {name}")

        with open(filepath, "rb") as f:
            checkpoint = pickle.load(f)

        print(f"체크포인트 로드: {name} (저장 시각: {checkpoint['timestamp']})")
        return checkpoint["state"]

    def list_checkpoints(self) -> list[dict]:
        """저장된 체크포인트 목록"""
        result = []
        for pkl_file in sorted(self.checkpoint_dir.glob("*.pkl")):
            with open(pkl_file, "rb") as f:
                cp = pickle.load(f)
            result.append({
                "name": cp["name"],
                "timestamp": cp["timestamp"],
                "size": pkl_file.stat().st_size,
                "metadata": cp.get("metadata", {}),
            })
        return result

    def delete(self, name: str) -> None:
        """체크포인트 삭제"""
        for ext in [".pkl", ".sha256"]:
            path = self.checkpoint_dir / f"{name}{ext}"
            if path.exists():
                path.unlink()
        print(f"체크포인트 삭제: {name}")

    @staticmethod
    def _compute_checksum(filepath: Path) -> str:
        h = hashlib.sha256()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                h.update(chunk)
        return h.hexdigest()


# 사용 예시
cp = Checkpoint("./checkpoints")

# 학습 상태 저장 시뮬레이션
training_state = {
    "epoch": 50,
    "loss": 0.0234,
    "accuracy": 0.987,
    "model_weights": [0.1, 0.2, 0.3, 0.4, 0.5],  # 실제로는 NumPy 배열
    "optimizer_state": {"lr": 0.001, "momentum": 0.9},
}

cp.save("training_epoch50", training_state,
        metadata={"model": "ResNet50", "dataset": "ImageNet"})

# 나중에 복원
restored_state = cp.load("training_epoch50")
print(f"복원된 에폭: {restored_state['epoch']}")
print(f"복원된 정확도: {restored_state['accuracy']}")

# 체크포인트 목록
for info in cp.list_checkpoints():
    print(f"  {info['name']}: {info['timestamp']} ({info['size']} bytes)")

---

고수 팁

1. pickle vs JSON 선택 기준

# pickle 사용: Python 전용, 모든 Python 객체, 빠름
# - 머신러닝 모델, numpy 배열, 복잡한 객체 구조
# - 같은 코드베이스 내 데이터 저장

# JSON 사용: 언어 무관, 사람이 읽을 수 있음, 안전
# - API 응답, 설정 파일, 로그
# - 다른 언어/시스템과 데이터 교환

# 선택 흐름
def choose_serialization(obj, cross_language=False, human_readable=False):
    if cross_language or human_readable:
        return "JSON (또는 msgpack)"
    elif hasattr(obj, "__array__"):  # numpy 배열
        return "joblib"
    elif callable(obj):  # 함수, 클로저
        return "dill"
    else:
        return "pickle"

2. pickle 파일에 버전 정보 포함

import pickle

CURRENT_VERSION = 2

def save_with_version(filepath: str, data: dict) -> None:
    versioned = {"version": CURRENT_VERSION, "data": data}
    with open(filepath, "wb") as f:
        pickle.dump(versioned, f)

def load_with_migration(filepath: str) -> dict:
    with open(filepath, "rb") as f:
        versioned = pickle.load(f)

    version = versioned.get("version", 1)
    data = versioned["data"]

    # 버전별 마이그레이션
    if version == 1:
        data = migrate_v1_to_v2(data)

    return data

def migrate_v1_to_v2(data: dict) -> dict:
    """v1 → v2 마이그레이션: 새 필드 추가"""
    data.setdefault("new_field", "default_value")
    return data

---

정리

모듈/라이브러리	특징	용도
pickle	Python 전용, 빠름, 보안 주의	내부 데이터 캐싱, 체크포인트
shelve	pickle 기반 딕셔너리 DB	간단한 영속적 저장
struct	C 구조체 형식 이진 처리	파일 포맷, 네트워크 프로토콜
joblib	NumPy/ML 모델 최적화	머신러닝 모델 저장
msgpack	크로스 언어, 빠름	다국어 데이터 교환
dill	람다/클로저 포함 직렬화	멀티프로세싱, 복잡한 함수

보안 수칙: pickle.loads()는 절대 신뢰할 수 없는 외부 데이터에 사용하지 마세요. 역직렬화 시 임의 코드가 실행될 수 있습니다.