4  PyArrow

pandas에서 데이터를 어떻게 저장하는지, 그리고 대용량 데이터를 처리하기 위해 이를 묶어내는 데이터 구조는 어떤 것들이 있는지 깊이 있게 살펴봅니다.

pandas 3.x 업데이트에서 거대하고 파괴적인 변화는 단연 Apache Arrow(및 그 Python 바인딩인 PyArrow)의 전면적인 도입입니다. pandas의 창시자인 웨스 맥키니(Wes McKinney)는 일찍이 2013년 자신의 블로그에 내가 pandas에 대해 싫어하는 10가지(10 Things I Hate About pandas)라는 유명한 글을 남긴 바 있습니다. 그는 이 글에서 기존 pandas가 가진 구조적 한계, 1) 무겁고 파편화된 문자열(String) 처리, 2) 결측치(NaN) 처리를 위한 억지스러운 강제 형변환, 3) 제한적인 멀티코어 활용 등을 심도깊게 논의했습니다.

특히 최근 pandas 공식 커뮤니티의 주요 제안인 PDEP-10(pandas Enhancement Proposal 10)과 PDEP-14(pandas Enhancement Proposal 14)를 기점으로, pandas 3.x에서는 기존의 NumPy object 문자열 타입을 대체하여 PyArrow 기반 문자열을 기본(default)으로 채택하는 등 Arrow를 적극적으로 도입하고 있습니다.

이 변화는 중요성이 높습니다. 데이터를 분석하는 입장에서 세부적인 사항까지 모두 알아야 할 필요는 없지만, CSV에서 parquet으로의 전환이 서서히 일어나고 있기 때문에, 데이터 처리와 분석의 내부 동작 원리를 이해하는 것은 중요합니다. 따라서 Arrow가 데이터 내부 구조를 어떻게 구현하는지 살펴봅시다.

4.1 논리와 물리적 구조

데이터 분석가는 보통 정수, 문자열, 타임스탬프와 같은 논리적인 타입(logical type)으로 데이터를 생각합니다. 하지만 하드웨어(CPU와 RAM)가 이를 초당 수십억 번씩 빠르게 연산하려면, 이 논리적 개념이 연속된 메모리 공간에 0과 1로 어떻게 배치될지 정의하는 물리 레이아웃(physical layout)이 필요합니다.

Arrow는 이 물리적 데이터 레이아웃을 엄격하게 규격화하여, CPU가 데이터를 한 번에 대량으로 가져와 처리하는 SIMD(Single Instruction Multiple Data) 연산의 효율을 극한으로 끌어올립니다.

4.1.1 Primitive 타입

pandas를 사용한 데이터 분석가들을 괴롭혔던 정수형(Integer) 컬럼에 결측치(NaN)가 하나라도 들어오면, 컬럼 전체가 실수형(Float)으로 바뀌는 현상이 있었습니다. 그 이유는 기반이 된 NumPy 배열이 자체적으로 결측치를 표현할 빈 공간(비트)을 마련해두지 않았기 때문입니다. pandas 문서도 데이터 타입에 따라 서로 다른 sentinel 값을 쓰며, 정수형은 NaN 때문에 float로 승격되는 문제가 있다고 설명합니다.

NEP 12의 “Boolean masks signalling missing values”, 결측치를 나타내기 위해 데이터와 병렬인 boolean mask를 두고, True/False로 유효값 여부를 관리합니다. 이 방식은 ’underlying data type의 값 비트 공간을 건드리지 않는다’는 점이 중요합니다. 값 저장 공간 안에 결측 전용 비트를 따로 넣는 것이 아니라, 마스크를 추가하는 설계입니다.

Arrow는 이 문제를 두 개의 핵심 버퍼(buffer, 연속된 메모리 덩어리)로 나누어 완벽하게 해결합니다.

  1. validity buffer: 각 값의 결측치(null) 여부를 단 1비트(0과 1) 단위로 빽빽하게 압축하여 저장합니다.

  2. values buffer: 실제 숫자 데이터를 연속된 바이트로 꽉 채워 저장합니다.

import pyarrow as pa
import struct

# 결측치(None)가 포함된 32비트 정수 배열 생성
arr_primitive = pa.array([1, 2, None, 4], type=pa.int32())

# 배열을 구성하는 하위 버퍼 추출
validity_buf, values_buf = arr_primitive.buffers()

print(
    f"Validity 버퍼 크기: {validity_buf.size} Bytes (결측치 유무만 1비트씩 압축 저장)"
)
print(
    f"Values 버퍼 크기: {values_buf.size} Bytes (실제 데이터 저장)"
)

# Values 버퍼를 Python 로우 바이트로 변환 후, C 구조체로 직접 디코딩해보기
raw_bytes = values_buf.to_pybytes()

# '<i'는 리틀 엔디안 4바이트 정수 포맷을 의미합니다.
decoded_values = [
    v[0] for v in struct.iter_unpack("<i", raw_bytes)
]
print(f"저수준 바이트 디코딩 결과: {decoded_values}")
Validity 버퍼 크기: 1 Bytes (결측치 유무만 1비트씩 압축 저장)
Values 버퍼 크기: 16 Bytes (실제 데이터 저장)
저수준 바이트 디코딩 결과: [1, 2, 0, 4]

위 결과를 보면 3번째 인덱스의 논리적 값은 None이지만, 메모리 디코딩 결과를 보면 0이라는 임의의 값(Garbage Value)이 채워진 채 공간을 차지하고 있습니다. 만약 결측치라고 해서 공간을 비워두거나 쪼개면 배열의 연속성이 깨져 연산 속도가 급감합니다. 따라서 데이터의 연속성을 끝까지 유지하되, 결측 여부는 가벼운 1비트짜리 배열(Validity Bitmap)에만 의존하여 판별함으로써 CPU 캐시를 100% 효율적으로 사용하는 것입니다.

4.1.2 가변 길이 타입

문자열(String)과 같은 가변 길이 타입은 치명적인 문제를 안고 있습니다. 단어마다 길이가 달라(‘A’ vs ‘Apple’) 메모리에 예쁘게 쌓을 수 없습니다. 기존 pandas의 object 타입은 Python 문자열 객체들이 메모리 여기저기에 흩어져 있고 그 위치(포인터)만 저장하는 방식을 썼습니다. 데이터가 커질수록 메모리 파편화가 일어나 속도가 느려집니다. Arrow는 데이터의 시작과 끝 위치를 가리키는 Offsets 버퍼를 추가하여 총 3개의 버퍼로 가변 길이를 관리합니다.

arr_string = pa.array(
    ["arrow", "메모리", None, "layout"], type=pa.string()
)
validity_buf, offsets_buf, values_buf = arr_string.buffers()

# Offsets 버퍼 해부 (각 문자열의 시작과 끝을 알려주는 이정표)
offset_bytes = offsets_buf.to_pybytes()
offsets = [
    o[0] for o in struct.iter_unpack("<i", offset_bytes)
]

print(f"Validity Bitmap 버퍼: {validity_buf}")
print(f"Values 버퍼: {values_buf}")
print(f"문자열 Offsets 리스트 (목차): {offsets}")
Validity Bitmap 버퍼: <pyarrow.Buffer address=0x2532fe200c0 size=1 is_cpu=True is_mutable=True>
Values 버퍼: <pyarrow.Buffer address=0x2532fe20100 size=20 is_cpu=True is_mutable=True>
문자열 Offsets 리스트 (목차): [0, 5, 14, 14, 20]

모든 문자열 텍스트는 Values 버퍼에 단 하나의 거대한 바이트 덩어리로 뭉쳐 저장됩니다. 특정 위치의 문자열을 찾을 때는 Offsets 버퍼에서 시작/끝 인덱스만 참조하므로, 포인터 탐색 없이 \(O(1)\)의 빠른 속도로 접근이 가능합니다.

PDEP-14(https://pandas.pydata.org/pdeps/0014-string-dtype.html)에 따라 pandas 3.x부터는 모든 문자열 데이터가 기본적으로 str 타입을 사용합니다.

4.2 Arrow 데이터 모델

물리적 버퍼를 바탕으로, PyArrow는 데이터를 블록처럼 조립하기 위한 4가지 주요 계층 객체를 제공합니다. 이는 pandas의 Series나 DataFrame이 내부적으로 어떻게 구성되는지 보여줍니다.

  1. Array (배열): 단일 타입의 연속된 버퍼 묶음입니다(작은 단위, Series에 해당).

  2. ChunkedArray (청크 배열): 여러 개의 Array를 논리적으로 기차처럼 하나로 엮은 것입니다. 배열 끝에 새로운 데이터를 추가할 때 전체 메모리를 복사(copy)하지 않고, 새로운 꼬리표(chunk)만 뒤에 덧붙여 비용을 극적으로 아낍니다.

  3. RecordBatch (배치): 스키마(컬럼명과 타입)를 공유하는 길이가 동일한 Array들의 2차원 모음입니다(작은 테이블 조각).

  4. Table (테이블): 여러 개의 ChunkedArray나 배치들을 엮어 만든 거대한 2차원 표 형식입니다. pandas의 DataFrame에 직접적으로 대응됩니다.

# 1. Array
arr = pa.array([1, 2, 3])

# 2. ChunkedArray(두 개의 Array를 '복사 없이' 기차처럼 이어붙임)
chunked = pa.chunked_array([arr, pa.array([4, 5])])
print(
    f"ChunkedArray 총 요소 수: {len(chunked)} / 내부 물리적 덩어리 수: {chunked.num_chunks}"
)

# 3. RecordBatch(하나의 완벽한 2차원 블록)
batch = pa.record_batch(
    [pa.array([1, 2]), pa.array(["A", "B"])],
    names=["id", "val"],
)
print(f"\nRecordBatch 구조:\n{batch}")

# 4. Table(배치들을 묶어 거대한 데이터프레임 생성)
table = pa.Table.from_batches([batch, batch])
print(f"Table 최종 차원: {table.shape}")
ChunkedArray 총 요소 수: 5 / 내부 물리적 덩어리 수: 2

RecordBatch 구조:
pyarrow.RecordBatch
id: int64
val: string
----
id: [1,2]
val: ["A","B"]
Table 최종 차원: (4, 2)

4.3 불변성과 영복사의 기적

Arrow 데이터를 다룰 때 중요하게 기억해야 할 대원칙은 한 번 생성된 메모리는 절대 덮어쓰거나 수정할 수 없다(불변성(immutability))는 것입니다. 값을 변경하거나 행을 지우려면, 새로운 배열을 복사해서 만들어야 합니다.

수정이 안 되면 비효율적이지 않을까 생각할 수 있습니다. 하지만 데이터가 절대 변하지 않는다는 확신이 있기 때문에, 수많은 스레드(Thread)나 프로세스가 데이터 엉킴을 방지하는 대기표(lock) 없이 안전하게 동일한 메모리를 동시에 읽을 수 있는 엄청난 이점이 생깁니다.

또한 데이터의 소유권(ownership) 개념이 적용됩니다. C++로 구현된 Arrow 메모리 풀에서, C Data Interface라는 규약을 활용하면 Python(pandas), R(dplyr), Rust(Polars), 초고속 SQL 엔진(DuckDB) 등 서로 다른 도구들이 데이터를 변환하거나 복사하지 않고 ’메모리 주소(주소값만 전달)’만 주고받으며 0초 만에 협력(zero-copy)할 수 있습니다.

4.4 효율적인 데이터 처리 파이프라인

4.4.1 용량이 큰 CSV 파일 처리

Kaggle에서 Instacart Market Basket Analysis를 분석한다고 가정해보겠습니다. 이 데이터셋은 약 713MB로, 메모리가 16GB인 환경에서 read_csv()로 충분히 읽을 수 있습니다. CSV 데이터는 데이터 교환에 불편한 점이 있습니다.

  1. 타입 정보가 없습니다: CSV는 순수 텍스트 형식이라 숫자, 문자열, 날짜 같은 데이터 타입을 저장할 수 없습니다. 파일을 읽을 때마다 pandas가 컬럼마다 타입을 추측해야 하는데, 이 과정에서 메모리가 낭비되거나 의도치 않은 타입으로 해석될 수 있습니다.

  2. 압축 효율이 낮습니다: CSV는 텍스트 기반이라 숫자 하나를 저장해도 문자열로 인코딩됩니다. 예를 들어 12345라는 숫자는 바이너리로 4바이트면 충분하지만, CSV에서는 5바이트 문자열로 저장됩니다. 게다가 반복되는 값이나 패턴이 있어도 압축할 방법이 없습니다.

  3. 부분 읽기가 불가능합니다: 컬럼 3개만 필요해도 CSV는 파일의 첫 바이트부터 끝까지 쉼표(,)와 줄바꿈을 일일이 찾아가며 전체를 스캔해야 합니다. 수백만 행의 데이터에서 특정 컬럼만 추출하려면 불필요한 나머지 90% 데이터도 모두 읽어야 합니다.

  4. 스키마 정보가 없습니다: 컬럼 이름, 기본값, nullable 여부 같은 메타데이터가 파일 자체에 포함되지 않습니다. 따라서 별도의 문서나 코드로 스키마를 관리해야 합니다.

이러한 문제를 해결하기 위해 바이너리 컬럼 형식인 Parquet 을 사용합니다. Parquet 은 Arrow 생태계에서 사용되는 바이너리 컬럼형(Columnar) 형식으로 쿼리에 필요한 특정 열만 지정해서 디스크에서 즉시 빼올 수 있습니다.

from pathlib import Path
import pandas as pd

data_dir = Path("../data/instacart-market-basket-analysis-csv")

# 1. Instacart CSV 데이터셋 전체 로드 및 조인
orders = pd.read_csv(data_dir / "orders.csv")
prior = pd.read_csv(data_dir / "order_products__prior.csv")
train = pd.read_csv(data_dir / "order_products__train.csv")
products = pd.read_csv(data_dir / "products.csv")
aisles = pd.read_csv(data_dir / "aisles.csv")
departments = pd.read_csv(data_dir / "departments.csv")

order_products = pd.concat([prior, train], ignore_index=True)

df_full = (
    order_products.merge(orders, on="order_id", how="left")
    .merge(products, on="product_id", how="left")
    .merge(aisles, on="aisle_id", how="left")
    .merge(departments, on="department_id", how="left")
)

mem_bytes = df_full.memory_usage(deep=True).sum()
mem_gb = mem_bytes / (1024**3)

print("기본 pandas 타입 전체 병합 (Full Merge) 결과")
print(f" - 병합 데이터 크기: {len(df_full):,}행 x {len(df_full.columns)}열")
print(f" - 최종 메모리 점유량: {mem_gb:.2f} GB ({mem_bytes / (1024**2):,.2f} MB)")

위 코드를 실행하면 아래와 유사한 결과를 확인할 수 있습니다.

기본 pandas 타입 전체 병합 (Full Merge) 결과
 - 병합 데이터 크기: 33,819,106행 x 15열
 - 최종 메모리 점유량: 5.45 GB (5,579.43 MB)

CSV는 텍스트 기반 형식으로, 특정 컬럼하나만 필요해도 컴퓨터는 파일의 첫 바이트부터 끝까지 쉼표(,)와 줄바꿈을 일일이 찾아가며 전체를 다 스캔해야 합니다.

이제 CSV 파일일을 Parquet으로 변경해보겠습니다. Parquet은 Arrow 생태계에서 사용되는 바이너리 컬럼형(columnar) 형식으로 쿼리에 필요한 특정 열만 지정해서 디스크에서 즉시 빼올 수 있습니다.

from pathlib import Path
import pyarrow.dataset as ds
import pyarrow.parquet as pq

data_dir = Path("../data/instacart-market-basket-analysis-csv")
parquet_dir = Path("../data/instacart-market-basket-analysis-parquet")
parquet_dir.mkdir(parents=True, exist_ok=True)

for csv_path in sorted(data_dir.glob("*.csv")):
    parquet_path = parquet_dir.joinpath(csv_path.with_suffix(".parquet").name)

    # PyArrow dataset으로 CSV 읽기 후 Parquet 저장 (Snappy 압축)
    dataset = ds.dataset(csv_path, format="csv")
    table = dataset.to_table()
    pq.write_table(table, parquet_path, compression="zstd")

    csv_mb = csv_path.stat().st_size / (1024 * 1024)
    parquet_mb = parquet_path.stat().st_size / (1024 * 1024)
    print(
        f"{csv_path.name} ({csv_mb:.2f} MB) -> {parquet_path.name} ({parquet_mb:.2f} MB)"
    )

위 코드를 사용하면 기존의 700MB가 넘던 CSV가 이제 120MB정도로 줄어든 것을 확인할 수 있습니다.

aisles.csv (0.00 MB) -> aisles.parquet (0.00 MB)
departments.csv (0.00 MB) -> departments.parquet (0.00 MB)
order_products__prior.csv (550.80 MB) -> order_products__prior.parquet (91.09 MB)
order_products__train.csv (23.54 MB) -> order_products__train.parquet (3.91 MB)
orders.csv (103.92 MB) -> orders.parquet (20.08 MB)
products.csv (2.07 MB) -> products.parquet (0.96 MB)

이제 변환된 Parquet 데이터셋(../data/instacart-market-basket-analysis-parquet)을 사용하여 기존 CSV 실험과 동일하게 전체 데이터를 로드하고 병합(full merge)하는 실험을 진행해 보겠습니다.

from pathlib import Path
import pandas as pd

data_dir = Path("../data/instacart-market-basket-analysis-parquet")

# 1. Instacart Parquet 데이터셋 전체 로드 및 조인
orders = pd.read_parquet(data_dir / "orders.parquet")
prior = pd.read_parquet(data_dir / "order_products__prior.parquet")
train = pd.read_parquet(data_dir / "order_products__train.parquet")
products = pd.read_parquet(data_dir / "products.parquet")
aisles = pd.read_parquet(data_dir / "aisles.parquet")
departments = pd.read_parquet(data_dir / "departments.parquet")

order_products = pd.concat([prior, train], ignore_index=True)

df_full_pq = (
    order_products.merge(orders, on="order_id", how="left")
    .merge(products, on="product_id", how="left")
    .merge(aisles, on="aisle_id", how="left")
    .merge(departments, on="department_id", how="left")
)

mem_bytes = df_full_pq.memory_usage(deep=True).sum()
mem_gb = mem_bytes / (1024**3)

print("Parquet 기반 pandas 타입 전체 병합 (Full Merge) 결과:")
print(f" - 병합 데이터 크기: {len(df_full_pq):,}행 x {len(df_full_pq.columns)}열")
print(f" - 최종 메모리 점유량: {mem_gb:.2f} GB ({mem_bytes / (1024**2):,.2f} MB)")

메모리 사용량은 크게 차이가 드러나지 않는데, 이는 pandas가 여전히 전체 데이터를 RAM에 로드하기 때문입니다. 하지만, CSV 파일 용량 대비 Parquet 파일 용량이 1/5 수준으로 줄었기 때문에 디스크 I/O 비용은 확실히 줄어든 것을 확인할 수 있습니다.

Parquet 기반 pandas 타입 전체 병합 (Full Merge) 결과
 - 병합 데이터 크기: 33,819,106행 x 15열
 - 최종 메모리 점유량: 5.45 GB (5,579.43 MB)

마지막으로 대용량 데이터를 GitHub 으로 배포하시는 분들을 위한 간단한 팁을 작성해보겠습니다. GitHub 정책에 따라 단일 파일이 100 메가가 넘어가면 Push 가 어렵습니다. 이럴 때는 Git LFS(large file storage)를 사용하거나, 데이터를 분할하여 업로드하는 방법을 고려해야 합니다. 이 중 우리는 분할하여 업로드하는 방법을 사용해보겠습니다.

from pathlib import Path
import pyarrow.dataset as ds

csv_path = Path("../data/instacart-market-basket-analysis-csv/order_products__prior.csv")
output_dir = Path("../data/instacart-market-basket-analysis-parquet-split")
output_dir.mkdir(parents=True, exist_ok=True)

# CSV를 PyArrow Table로 로드
table = ds.dataset(csv_path, format="csv").to_table()

# 전체 3,240만 행 중 약 30MB에 해당하는 8,500,000행 단위로 분할
rows_per_file = 8_500_000

ds.write_dataset(
    table,
    base_dir=output_dir,
    format="parquet",
    max_rows_per_file=rows_per_file,
    max_rows_per_group=rows_per_file,  # max_rows_per_group <= max_rows_per_file 조건
    basename_template="order_products_prior_part_{i}.parquet",
    existing_data_behavior="overwrite_or_ignore"
)

분할된 여러 Parquet 파일을 하나의 논리적 데이터셋으로 취급합니다.

from pathlib import Path
import pyarrow.dataset as ds

data_dir = Path("../data/instacart-market-basket-analysis-parquet-split")

# 분할된 여러 Parquet 파일을 하나의 Dataset 으로 인식
dataset = ds.dataset(
    data_dir,
    format="parquet",
    partitioning=None  # 파티션 없이 단일 디렉터리로 처리
)

# 전체 스캔 (내부적으로 여러 파일을 순차 읽음)
scanner = dataset.scanner()
table = scanner.to_table()
df = table.to_pandas()

print(f"총 행 수: {len(df):,}")

4.4.2 파티션 가지치기와 행 그룹 스킵

파티션 가지치기는 year=2026/month=07 같은 디렉터리 구조를 보고, 필터와 맞지 않는 파티션 전체를 dataset 구성 단계에서 제외하는 방식입니다. 반면 행 그룹 스킵(row-group skipping)은 같은 파일 안에서도 각 행 그룹의 min/max 통계를 보고, 조건에 맞을 가능성이 없는 블록을 읽기 전에 건너뜁니다.

from pathlib import Path
import pandas as pd
import pyarrow.dataset as ds
import pyarrow.compute as pc

data_dir = Path(
    "../data/instacart-market-basket-analysis-parquet"
)

# 시나리오: "평일 아침 6시~12시 주문만 분석하고 싶다"
# order_dow: 0-4(평일), order_hour_of_day: 6-11

# 1. 전체 데이터 로드 (비교용)
orders_full = pd.read_parquet(data_dir / "orders.parquet")
filtered_pandas = orders_full[
    (orders_full["order_dow"] < 5)
    & (orders_full["order_hour_of_day"].between(6, 11))
]

# 2. PyArrow Dataset + 필터 적용 (파티션 가지치기 + Row-group Skipping)
dataset = ds.dataset(
    data_dir / "orders.parquet", format="parquet"
)

# Predicate pushdown: 필터 조건을 스캔 단계에서 적용
scanner = dataset.scanner(
    filter=(
        (pc.field("order_dow") < 5)
        & (pc.field("order_hour_of_day") >= 6)
        & (pc.field("order_hour_of_day") <= 11)
    )
)

table_filtered = scanner.to_table()
df_arrow = table_filtered.to_pandas()
df_arrow
order_id user_id ... order_hour_of_day days_since_prior_order
0 2539329 1 ... 8 NaN
1 2398795 1 ... 7 15.0
2 2254736 1 ... 7 29.0
... ... ... ... ... ...
837758 3154581 206209 ... 11 NaN
837759 688306 206209 ... 10 30.0
837760 1854736 206209 ... 10 30.0

837761 rows × 7 columns

아래와 같이 파티셔닝된 instacart-orders-parquet 디렉터리에서, order_dow=5/와 order_dow=6/는 order_dow < 5 조건에 맞지 않아 스캔 대상에서 제외됩니다. 이와 같이, 디렉터리 수준에서 필터와 맞지 않는 파티션 전체를 제외하는 기법이 파티션 가지치기(partition Pruning)입니다.

instacart-orders-parquet/
├── order_dow=0/
├── order_dow=1/
├── ...
├── order_dow=5/  ← 주말 (스캔 대상 제외)
└── order_dow=6/  ← 주말 (스캔 대상 제외)

4.4.3 스캐너와 to_batches()를 이용한 스트리밍 집계 연산

거대한 데이터를 to_table()이나 read_parquet()으로 한 번에 읽으면 컴퓨터가 멈춥니다(out of memory). 대신 to_batches()를 통해 데이터를 쪼개서 흘려보내며(Streaming) 필요한 연산만 누적해야 합니다. Dataset → Scanner → RecordBatch → Compute로 이어지는 실무 패턴을 작성해 보겠습니다.

import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.dataset as ds

# 1. Instacart orders.parquet 데이터셋 로드 (Parquet 포맷 지정)
dataset = ds.dataset(
    "../data/instacart-market-basket-analysis-parquet/orders.parquet",
    format="parquet",
)

# 2. Scanner 구성: (디스크 I/O 최소화의 핵심!)
scanner = dataset.scanner(
    columns=["order_dow", "order_hour_of_day"],
    filter=(pc.field("eval_set") == "prior"),
)

# 3. 누적 집계 상태를 저장할 작은 Python 딕셔너리
total_hours = {}
total_orders = {}

# 4. to_batches() 스트리밍: 대용량 데이터라도 설정한 청크 크기(수 MB)만 메모리에 올라옵니다.
for batch in scanner.to_batches():
    # RecordBatch를 PyArrow Table로 변환하여 C++ 엔진 기반 고속 group_by 연산 수행
    tbl = pa.Table.from_batches([batch])
    grouped = tbl.group_by("order_dow").aggregate(
        [
            ("order_hour_of_day", "sum"),
            ("order_hour_of_day", "count"),
        ]
    )

    # 쪼개진 배치별 집계 결과를 Python 딕셔너리에 계속 누적합니다.
    for i in range(grouped.num_rows):
        dow = grouped.column("order_dow")[i].as_py()
        hour_sum = grouped.column("order_hour_of_day_sum")[
            i
        ].as_py()
        order_cnt = grouped.column(
            "order_hour_of_day_count"
        )[i].as_py()

        total_hours[dow] = (
            total_hours.get(dow, 0) + hour_sum
        )
        total_orders[dow] = (
            total_orders.get(dow, 0) + order_cnt
        )

print("스트리밍 집계 완료")
dow_names = [
    "일요일",
    "월요일",
    "화요일",
    "수요일",
    "목요일",
    "금요일",
    "토요일",
]
for dow in sorted(total_hours.keys()):
    avg_hour = total_hours[dow] / total_orders[dow]
    day_name = (
        dow_names[dow] if 0 <= dow < 7 else f"요일 {dow}"
    )
    print(
        f" - {day_name}({dow}): 총 {total_orders[dow]:,}건 / 평균 {avg_hour:.2f}시"
    )
스트리밍 집계 완료
 - 일요일(0): 총 557,772건 / 평균 13.57시
 - 월요일(1): 총 556,705건 / 평균 13.16시
 - 화요일(2): 총 441,955건 / 평균 13.46시
 - 수요일(3): 총 412,400건 / 평균 13.52시
 - 목요일(4): 총 401,212건 / 평균 13.58시
 - 금요일(5): 총 425,982건 / 평균 13.36시
 - 토요일(6): 총 418,848건 / 평균 13.52시

이 방법을 사용하면 데이터의 용량이 크더라도 Python이 실제로 사용하는 메모리는 적게 유지됩니다. 그래서 메모리가 부족한 환경에서도 안전하게 최종 결과를 얻을 수 있습니다.

4.5 pandas와 Arrow의 통합

Arrow 도입에 따른 혼란을 줄이고, 수많은 유저들의 기존 코드가 깨지는 것을 막기 위해서 pandas는 점진적인 통합을 선택했습니다.

4.5.1 파서 엔진(engine)과 저장 백엔드(dtype_backend)의 완전한 구분

pandas의 read_csv() 함수에는 초/중급 사용자들이 헷갈려하는 두 가지 옵션이 있습니다. 이 둘은 완전히 독립적인 축으로 작동합니다.

  1. engine="pyarrow": 데이터를 하드디스크에서 읽고 구문 분석(parsing)하는 ’작업자(실행 엔진)’입니다. C++ 기반 멀티스레딩 파싱을 사용하여 읽는 속도가 매우 빠릅니다.

  2. dtype_backend="pyarrow": 파싱이 완료된 데이터를 메모리에 담아둘 ’그릇의 형태(저장 방식)’입니다. 이를 지정해야만 메모리상에 ArrowDtype을 사용하는 최적화된 컬럼이 최종 생성됩니다.

import pandas as pd
import io

csv_data = io.StringIO(
    "id,value\n1,10.5\n2,20.1\n3,"
)  # 3번 값 누락(NaN)

# 1. 파싱만 PyArrow로 빠르게 하고, 저장은 전통적인 NumPy 방식으로
df_np = pd.read_csv(csv_data, engine="pyarrow")
print("NumPy 저장소 사용(기본):", df_np["value"].dtype)

csv_data.seek(0)  # 스트림 커서 초기화

# 2. 파싱 엔진과 최종 저장소 모두 PyArrow를 명시적으로 사용
df_pa = pd.read_csv(
    csv_data, engine="pyarrow", dtype_backend="pyarrow"
)
print("Arrow 저장소 명시 사용:", df_pa["value"].dtype)
NumPy 저장소 사용(기본): float64
Arrow 저장소 명시 사용: double[pyarrow]

pandas 3.x에서 숫자 컬럼은 dtype_backend를 별도로 명시하지 않으면 여전히 기존 호환성을 위해 NumPy 백엔드를 사용합니다. (단, 문자열 데이터는 PDEP-14에 의해 자동으로 PyArrow로 처리됩니다.) 결측치가 들어가도 원본 정수형(int64[pyarrow])이 실수형(float64)으로 오염되지 않고 그대로 보존되는 것은 dtype_backend를 켰을 때 얻을 수 있는 큰 장점 중 하나입니다.

4.5.2 폴백(Fallback) 메커니즘 주의

PyArrow 백엔드(dtype_backend="pyarrow")가 적용된 데이터프레임에서 .mean(), .sum() 등의 연산을 수행할 때, pandas는 내부적으로 PyArrow의 C++ Compute API로 이를 직접 실행하려 시도합니다. 성공하면 변환 없이 초고속 벡터 연산이 수행됩니다.

하지만 GitHub 이슈 트래커 등에서 자주 보고되듯, PyArrow가 아직 지원하지 않는 복잡한 Python 특화 통계 함수나, 커스텀 .apply(lambda x: ...)를 실행하면 어떻게 될까요? pandas는 에러를 내는 대신 몰래 데이터를 무거운 NumPy 배열 형식으로 임시 복사(변환)한 뒤 연산을 수행하고, 결과를 다시 Arrow 형식으로 되돌려놓습니다. 이를 폴백(fallback)이라고 부릅니다. 이 과정에서 막대한 메모리 복사 비용과 속도 저하(out of memory)가 발생할 수 있으므로, 대규모 데이터 처리 시 갑자기 속도가 비정상적으로 느려진다면 내가 쓴 코드가 폴백을 유발했는지 의심해 보아야 합니다.

4.5.3 조건부 영복사 검증

PyArrow Table과 pandas DataFrame 간의 상호 변환(.to_pandas())은 종종 메모리 복사가 전혀 없는 영복사(zero-copy) 마법으로 홍보되지만, 항상 보장되는 것은 아닙니다.

타입 호환성, 결측치(null) 여부, 그리고 특히 앞서 배운 ChunkedArray의 내부 청크 조각이 잘게 쪼개져 있는 경우, 연속된 메모리 레이아웃을 요구하는 형태로 변환하기 위해 결국 값 전체를 복사(copy)해야만 하는 경우가 빈번합니다.

import pyarrow as pa

table = pa.table({'a': pa.array([1, 2, 3])})

# types_mapper 지정 시: 정확한 Arrow 백엔드 기반 Series를 생성하여 Zero-Copy 성공률이 극대화됨
df_arrow = table.to_pandas(types_mapper=pd.ArrowDtype)

# 지정하지 않을 시: 타입 호환성을 억지로 맞추기 위해 NumPy 변환 및 메모리 대규모 복사가 발생할 수 있음
df_numpy = table.to_pandas()

결론적으로, pandas 3.x 환경에서 PyArrow 생태계의 압도적인 퍼포먼스를 온전히 활용하려면 dtype_backend="pyarrow"를 명시적으로 지정하는 습관을 들이고, 의도치 않게 발생하는 폴백(fallback) 연산을 경계하며, 변환 과정에서 영복사(zero-copy)가 깨지는 조건들을 정확히 이해해야 합니다. 이 원리들을 체득한다면 아무리 방대한 데이터라도 가볍게 요리하는 마스터 분석가로 도약할 수 있을 것입니다.