Streaming / tempfile / mmap — 읽기/쓰기 너머

Streaming — 파일이 RAM 보다 클 때

디폴트 file 객체 iterator 가 한 줄씩 줘. 대부분 텍스트 로그엔 충분. 바이너리 stream 엔 f.read(chunk_size) 루프 또는 walrus 연산자. 핵심 — 10GB 파일에 인자 없는 .read() 호출 하지 마, RAM 10GB 없으면.

tempfile — scratch 공간 필요할 때

tempfile.NamedTemporaryFile() 가 시스템 temp 디렉토리에 유니크 이름 파일 + 자동 정리. tempfile.TemporaryDirectory() 가 디렉토리 + 컨텍스트 종료 시 재귀 정리. 처리 파이프라인, 테스트 fixture, path 약속 없이 디스크 공간 필요한 모든 거에 사용.

mmap — 파일을 메모리로

mmap 이 파일을 프로세스 가상 메모리에 매핑. bytearray 처럼 접근, OS 가 페이징 처리 — 실제 만진 부분만 로드. 멀티-GB 파일 random-access 읽기에 seek+read 보다 극적으로 빠름. 쓰기엔 mmap-수정이 효율적으로 flush 가능.

StringIO / BytesIO — 메모리의 파일

io 모듈 — StringIO 와 BytesIO 가 메모리 backed file 같은 객체. 테스트 (함수에 fake 파일 줌), 안 그러면 파일로 갈 출력 캡쳐, 한 번 flush 전 프로그램으로 파일 내용 빌드에 유용. 진짜 file 객체와 같은 메서드 지원.

War Story: cwkPippa 의 세션 로그가 JSONL — 줄당 하나의 JSON 객체. Append-only 디자인 = 전체 대화 절대 버퍼링 안 하고 일어나는 대로 이벤트 stream-write. 복구 코드가 줄 단위로 읽음. 맞는 일에 맞는 모양.

Code

큰 바이너리 파일 streaming·python

import io

# 가짜 파일 (데모용 메모리 backed)
big = io.BytesIO(b"xyz" * 1_000_000)        # 3MB

CHUNK = 4096
big.seek(0)
total = 0
while chunk := big.read(CHUNK):
    total += len(chunk)
print("읽은 바이트:", total)

# 또는 iterator 패턴
big.seek(0)
import functools
for chunk in iter(functools.partial(big.read, CHUNK), b""):
    pass
# iter() + sentinel — big.read(CHUNK) 가 b'' 반환할 때까지 호출

tempfile — 자동 정리되는 scratch 공간·python

import tempfile
from pathlib import Path

# Named 파일 — 닫힐 때 자동 삭제
with tempfile.NamedTemporaryFile(suffix=".txt", delete=True) as tf:
    print("path:", tf.name)
    tf.write(b"hello world")
    tf.flush()
    # 여기 파일 디스크에 있음
    print("크기:", Path(tf.name).stat().st_size)
# 파일 사라짐

# 디렉토리 — 재귀 정리
with tempfile.TemporaryDirectory() as td:
    p = Path(td)
    (p / "a.txt").write_text("a", encoding="utf-8")
    (p / "b.txt").write_text("b", encoding="utf-8")
    print("내용:", list(p.iterdir()))
# 디렉토리 + 내용 사라짐

mmap — random access 위한 file-as-memory·python

import mmap
import tempfile
from pathlib import Path

# 샘플 파일
p = Path("/tmp/mmap_demo.txt")
p.write_bytes(b"hello world hello pippa hello universe")

with open(p, "r+b") as f:
    with mmap.mmap(f.fileno(), 0) as mm:
        # 검색
        idx = mm.find(b"pippa")
        print("발견 위치:", idx)

        # 슬라이스 — bytes 객체처럼
        print(mm[0:5])              # b'hello'

        # 변경 — 파일에 write 통과 (WRITE 권한)
        mm[0:5] = b"HOWDY"

# 변경 지속 확인 위해 다시 읽기
print(p.read_bytes()[:30])
p.unlink()

StringIO — 테스트용 fake 파일·python

import io
import csv

# 진짜 파일 안 쓰고 CSV 읽는 함수 테스트
fake = io.StringIO('''name,age
alice,30
bob,25
''')

rows = list(csv.DictReader(fake))
print(rows)
# [{'name': 'alice', 'age': '30'}, {'name': 'bob', 'age': '25'}]

# 바이너리도 같음
binary = io.BytesIO(b"\xDE\xAD\xBE\xEF")
print(binary.read(2))                # b'\xde\xad'

Exercise

(a) tempfile.TemporaryDirectory 만들고 안에 pathlib 로 3 파일 쓰기. 디렉토리 내용 출력. with 블록 후 디렉토리 사라짐 확인. (b) 1MB 파일 만들기 (os.urandom 또는 random 바이트). 3 방식으로 읽기 — 한 번에 다, 4KB chunk, mmap 통해. 셋 다 같은 내용 확인 (예 — bytes 의 SHA256 계산).

Streaming / tempfile / mmap — 읽기/쓰기 너머

Streaming — 파일이 RAM 보다 클 때

tempfile — scratch 공간 필요할 때

mmap — 파일을 메모리로

StringIO / BytesIO — 메모리의 파일

Code

External links

Exercise

Progress

댓글 0