청크 간 점진적 집계

점진적 집계가 필요한 이유

행 개수 세기 및 평균 계산하기

import pandas as pd

total_sum = 0.0
total_count = 0

for chunk in pd.read_csv('transactions.csv', chunksize=100000):
    total_sum += chunk['amount'].sum()
    total_count += chunk['amount'].notna().sum()

grand_mean = total_sum / total_count
print(f'Rows processed: {total_count:,}')
print(f'Grand mean: {grand_mean:.4f}')

점진적으로 최솟값과 최댓값 계산하기

import pandas as pd

global_min = float('inf')
global_max = float('-inf')

for chunk in pd.read_csv('prices.csv', chunksize=50000):
    chunk_min = chunk['price'].min()
    chunk_max = chunk['price'].max()
    if chunk_min < global_min:
        global_min = chunk_min
    if chunk_max > global_max:
        global_max = chunk_max

print(f'Price range: {global_min} to {global_max}')

점진적으로 빈도 세기

import pandas as pd

freq = pd.Series(dtype='int64')

for chunk in pd.read_csv('orders.csv',
                         chunksize=100000,
                         usecols=['category']):
    chunk_counts = chunk['category'].value_counts()
    freq = freq.add(chunk_counts, fill_value=0)

# Final sorted frequency table
print(freq.sort_values(ascending=False).head(10))

점진적으로 GroupBy 집계하기

import pandas as pd

partials = []
for chunk in pd.read_csv('sales.csv',
                         chunksize=100000,
                         usecols=['region', 'product', 'revenue']):
    p = chunk.groupby(['region', 'product'])['revenue'].sum()
    partials.append(p)

final = (
    pd.concat(partials)
    .groupby(level=['region', 'product'])
    .sum()
    .sort_values(ascending=False)
)
print(final.head(10))

분산을 점진적으로 계산하기(Welford 방법)

import pandas as pd
import numpy as np

# Simple two-pass approach using stored chunk stats
chunk_stats = []
for chunk in pd.read_csv('data.csv',
                         chunksize=100000,
                         usecols=['value']):
    n = chunk['value'].count()
    mean = chunk['value'].mean()
    var = chunk['value'].var(ddof=1)
    chunk_stats.append((n, mean, var))

# Combine: use pooled variance formula
total_n = sum(s[0] for s in chunk_stats)
total_mean = sum(s[0]*s[1] for s in chunk_stats) / total_n
pooled_var = sum((s[0]-1)*s[2] + s[0]*(s[1]-total_mean)**2
                 for s in chunk_stats) / (total_n - 1)
print(f'Grand variance: {pooled_var:.4f}')

점진적 히스토그램 만들기

import pandas as pd
import numpy as np

# Decide bin edges from a sample
sample = pd.read_csv('amounts.csv', nrows=5000)
bins = np.linspace(sample['amount'].min(),
                   sample['amount'].max(), 21)
counts = np.zeros(len(bins) - 1, dtype='int64')

for chunk in pd.read_csv('amounts.csv',
                         chunksize=100000,
                         usecols=['amount']):
    chunk_counts, _ = np.histogram(
        chunk['amount'].dropna(), bins=bins
    )
    counts += chunk_counts

print('Histogram counts:', counts[:5], '...')

고유값을 근사적으로 추적하기

import pandas as pd

unique_ids = set()
for chunk in pd.read_csv('events.csv',
                         chunksize=100000,
                         usecols=['user_id']):
    unique_ids.update(chunk['user_id'].dropna().unique())

print(f'Distinct user IDs: {len(unique_ids):,}')
# Warning: the set may grow large for high-cardinality columns

장시간 실행 중 진행 상황 보고하기

import pandas as pd
import time

chunksize = 100000
start = time.time()
rows_processed = 0

for i, chunk in enumerate(pd.read_csv('big.csv',
                                       chunksize=chunksize)):
    rows_processed += len(chunk)
    # Report every 10 chunks
    if (i + 1) % 10 == 0:
        elapsed = time.time() - start
        rate = rows_processed / elapsed
        print(f'Chunk {i+1}: {rows_processed:,} rows '
              f'@ {rate/1000:.0f}k rows/sec')

print(f'Total: {rows_processed:,} rows in {time.time()-start:.1f}s')

집계 전에 필터링하기

import pandas as pd

partials = []
for chunk in pd.read_csv('orders.csv',
                         chunksize=100000,
                         parse_dates=['order_date']):
    # Filter early: only 2024 orders
    mask = chunk['order_date'].dt.year == 2024
    filtered = chunk.loc[mask, ['category', 'revenue']]

    if len(filtered) > 0:
        p = filtered.groupby('category')['revenue'].sum()
        partials.append(p)

if partials:
    result = pd.concat(partials).groupby(level=0).sum()
    print(result)

중간 결과 저장하기

import pandas as pd
import os

CHECKPOINT = 'checkpoint.csv'
running_total = 0.0
running_count = 0

# Resume from checkpoint if it exists
if os.path.exists(CHECKPOINT):
    ckpt = pd.read_csv(CHECKPOINT)
    running_total = ckpt['total'].iloc[0]
    running_count = int(ckpt['count'].iloc[0])
    print(f'Resuming from checkpoint: {running_count:,} rows')

for chunk in pd.read_csv('huge.csv', chunksize=100000):
    running_total += chunk['value'].sum()
    running_count += len(chunk)

# Save checkpoint
pd.DataFrame({'total': [running_total],
              'count': [running_count]}).to_csv(CHECKPOINT, index=False)
print(f'Final mean: {running_total / running_count:.4f}')

빠른 확인

강의 요약