대용량 파일 청크(Chunk) 스트리밍과
메모리 폭발(OOM) 방지 기법

import pandas as pd
import time

def process_large_csv_by_chunks(csv_path: str, chunk_size: int = 100_000):
    """10만 행 단위로 청크를 나누어 읽으며 총합과 카운트를 스트리밍 누적"""
    start_time = time.time()
    
    # 누적기 변수
    total_sales = 0.0
    total_orders = 0
    category_counter = {}
    
    print(f"🚀 청크 스트리밍 처리 시작 (청크 크기: {chunk_size:,} 행)")
    
    # chunksize 인자를 지정하면 DataFrame 대신 TextFileReader 제너레이터 반환
    chunk_reader = pd.read_csv(
        csv_path, 
        chunksize=chunk_size,
        usecols=['order_id', 'category', 'amount'] # 불필요한 열 제외 (I/O 절약)
    )
    
    for i, chunk in enumerate(chunk_reader, start=1):
        # 1. 청크별 부분 집계 수행 (RAM 점유는 청크 1개 분량에 불과함)
        chunk_sum = chunk['amount'].sum()
        chunk_count = len(chunk)
        
        total_sales += chunk_sum
        total_orders += chunk_count
        
        # 카테고리별 빈도수 누적
        for cat, cnt in chunk['category'].value_counts().items():
            category_counter[cat] = category_counter.get(cat, 0) + cnt
            
        print(f"  [Chunk {i:03d}] 처리 완료: 누적 주문 {total_orders:,} 건 (현재 메모리 안정적 유지)")
        
    elapsed = time.time() - start_time
    print(f"\n✅ 전체 집계 완료 (소요 시간: {elapsed:.2f}초)")
    print(f"💰 총 매출액: ₩{total_sales:,.0f} | 총 주문 수: {total_orders:,} 건")
    print(f"📊 상위 3대 카테고리: {sorted(category_counter.items(), key=lambda x: x[1], reverse=True)[:3]}")

# 실행 예시:
# process_large_csv_by_chunks('large_ecommerce_orders.csv', chunk_size=200_000)

import pandas as pd
import numpy as np

def auto_reduce_mem_usage(df: pd.DataFrame, verbose: bool = True) -> pd.DataFrame:
    """모든 컬럼의 최솟값/최댓값을 검사하여 가장 작은 비트의 타입으로 자동 다운캐스팅"""
    numerics = ['int16', 'int32', 'int64', 'float16', 'float32', 'float64']
    start_mem = df.memory_usage(deep=True).sum() / 1024**2
    
    for col in df.columns:
        col_type = df[col].dtype
        
        # 1. 수치형 데이터 다운캐스팅
        if col_type in numerics:
            c_min = df[col].min()
            c_max = df[col].max()
            if str(col_type).startswith('int'):
                if c_min > np.iinfo(np.int8).min and c_max < np.iinfo(np.int8).max:
                    df[col] = df[col].astype(np.int8)
                elif c_min > np.iinfo(np.int16).min and c_max < np.iinfo(np.int16).max:
                    df[col] = df[col].astype(np.int16)
                elif c_min > np.iinfo(np.int32).min and c_max < np.iinfo(np.int32).max:
                    df[col] = df[col].astype(np.int32)
            else:
                if c_min > np.finfo(np.float32).min and c_max < np.finfo(np.float32).max:
                    df[col] = df[col].astype(np.float32)
                    
        # 2. 객체형(문자열) 카테고리화: 고유값 개수가 전체의 50% 미만이면 Category 변환
        elif col_type == object:
            if df[col].nunique() / len(df[col]) < 0.5:
                df[col] = df[col].astype('category')
                
    end_mem = df.memory_usage(deep=True).sum() / 1024**2
    if verbose:
        reduction = 100 * (1 - end_mem / start_mem)
        print(f"📉 [메모리 다이어트 결과] {start_mem:.2f} MB ➡️ {end_mem:.2f} MB ({reduction:.1f}% 절감)")
    return df

import pandas as pd
import pyarrow.parquet as pq
import pyarrow as pa
import os

# 청크 단위로 CSV를 읽어 단일 Parquet 파일로 압축 스트리밍 기록
def convert_csv_to_parquet_stream(csv_file: str, parquet_file: str, chunk_size: int = 250_000):
    writer = None
    
    for chunk in pd.read_csv(csv_file, chunksize=chunk_size):
        # 메모리 다이어트 적용
        chunk = auto_reduce_mem_usage(chunk, verbose=False)
        
        # Pandas -> Apache Arrow 테이블 변환 (Zero-Copy 지원)
        table = pa.Table.from_pandas(chunk)
        
        # 첫 번째 청크일 때 파케이 작성기 초기화 (Zstandard 압축 적용)
        if writer is None:
            writer = pq.ParquetWriter(parquet_file, table.schema, compression='zstd')
            
        writer.write_table(table)
        print(f"  ⚡ {len(chunk):,} 행 파케이 스트림 압축 기록 완료")
        
    if writer:
        writer.close()
        
    csv_size = os.path.getsize(csv_file) / (1024**2)
    pq_size = os.path.getsize(parquet_file) / (1024**2)
    print(f"\n📦 [포맷 변환 완료] CSV {csv_size:.1f}MB ➡️ Parquet {pq_size:.1f}MB (압축률: {(1 - pq_size/csv_size)*100:.1f}%)")

import polars as pl

# scan_csv는 즉시 실행되지 않고 '실행 계획(LazyFrame)'만 생성합니다.
lazy_plan = (
    pl.scan_csv("massive_click_stream.csv")
    .filter(pl.col("status_code") == 200)
    .filter(pl.col("dwell_time_sec") > 10)
    .group_by("referrer_domain")
    .agg([
        pl.col("user_id").n_unique().alias("unique_visitors"),
        pl.col("dwell_time_sec").mean().alias("avg_dwell_time")
    ])
    .sort("unique_visitors", descending=True)
    .limit(10)
)

# collect(streaming=True)를 호출하면 메모리 한계를 넘지 않도록
# Polars 내부의 스트리밍 엔진이 청크별로 고속 연산을 처리합니다!
top_referrers = lazy_plan.collect(streaming=True)

print("⚡ [Polars Streaming Query 결과]")
print(top_referrers)

import pandas as pd
import numpy as np
import pyarrow.parquet as pq
import pyarrow as pa
import time
import os

class BigDataStreamCleaner:
    def __init__(self, chunk_size: int = 150_000, compression: str = 'zstd'):
        self.chunk_size = chunk_size
        self.compression = compression
        self.total_processed = 0
        self.total_dropped_nulls = 0
        
    def clean_chunk(self, chunk: pd.DataFrame) -> pd.DataFrame:
        """단일 청크 데이터 정제 및 타입 최적화"""
        # 1. 결측치 제거
        before_len = len(chunk)
        chunk = chunk.dropna(subset=['order_id', 'amount'])
        self.total_dropped_nulls += (before_len - len(chunk))
        
        # 2. 음수 이상치 필터링
        chunk = chunk[chunk['amount'] > 0]
        
        # 3. 날짜형 변환
        if 'order_date' in chunk.columns:
            chunk['order_date'] = pd.to_datetime(chunk['order_date'])
            
        # 4. 수치형 다운캐스팅
        for col in chunk.select_dtypes(include=['int64']).columns:
            chunk[col] = pd.to_numeric(chunk[col], downcast='integer')
        for col in chunk.select_dtypes(include=['float64']).columns:
            chunk[col] = pd.to_numeric(chunk[col], downcast='float')
            
        # 5. 저빈도 문자열 범주형 변환
        for col in chunk.select_dtypes(include=['object']).columns:
            if chunk[col].nunique() / len(chunk[col]) < 0.4:
                chunk[col] = chunk[col].astype('category')
                
        return chunk

    def run_pipeline(self, input_csv: str, output_parquet: str):
        print(f"🎬 [ETL 파이프라인 가동] {input_csv} ➡️ {output_parquet}")
        start_t = time.time()
        writer = None
        
        for i, chunk in enumerate(pd.read_csv(input_csv, chunksize=self.chunk_size), start=1):
            cleaned = self.clean_chunk(chunk)
            table = pa.Table.from_pandas(cleaned)
            
            if writer is None:
                writer = pq.ParquetWriter(output_parquet, table.schema, compression=self.compression)
            writer.write_table(table)
            
            self.total_processed += len(cleaned)
            print(f"  [Chunk {i:03d}] 정제 완료 | 누적 {self.total_processed:,} 건 저장됨")
            
        if writer:
            writer.close()
            
        elapsed = time.time() - start_t
        print(f"\n🎉 [성공] 총 {self.total_processed:,} 건 정제 완료 (제거된 결측치: {self.total_dropped_nulls:,}건)")
        print(f"⏱️ 총 소요 시간: {elapsed:.2f}초 | 출력 파일 크기: {os.path.getsize(output_parquet)/(1024**2):.1f} MB")

if __name__ == '__main__':
    pipeline = BigDataStreamCleaner(chunk_size=100_000)
    print("클래스 준비 완료. 실제 데이터 파일 경로를 지정하여 실행하세요!")