Posted inPython

PythonとPyArrowによるParquetファイル処理：読み書きの実装と本番運用の最適化ノウハウ

Python tutorial - IT technology blog
Python tutorial - IT technology blog

午前3時の障害と12GBのCSVデータがもたらした課題

私たちのシステムには、毎朝3時に実行されるcronジョブがあります。その役割は、Webサーバー群からアクセスログやイベントトラッキングデータを集約することでした。半年ほど前、日次で蓄積されるログの量が未加工のCSV形式で12GBを突破しました。

その結果、問題はすぐに表面化しました。おなじみの pd.read_csv() を実行するスクリプトが、わずか4分でVPSの16GBメモリ（RAM）を完全に食いつぶしたのです。CPU使用率は100%に張り付き、LinuxカーネルのOOM Killerが発動してプロセスを即座に強制終了しました。

当時、選択肢は2つありました。1つはRAMを32GBや64GBにスケールアップすることですが、これにはサーバー費用が毎月40〜60ドル余計にかかります。もう1つは、ストレージのデータフォーマットと読み込みパイプラインを一から再構築することでした。私は後者を選択し、すべてのデータを Apache Parquet に移行し、メインの処理エンジンとして PyArrow を採用しました。

Parquetは列指向（Columnar storage）フォーマットであり、行ごとにベタ書きされるCSVとは根本的に構造が異なります。Parquetが圧倒的に優れている技術的メリットは次の3点です。

  • 圧倒的なデータ圧縮率： 同一列内の値はすべて同じデータ型を持ちます。そのため、SnappyやZSTDといったアルゴリズムによって、プレーンテキストに比べてはるかに効率的にデータをグループ化・圧縮できます。12GBのCSVファイルが、わずか1.4GB〜1.8GB程度にまで圧縮され、ディスク容量を85%以上削減できました。
  • プロジェクションと述語プッシュダウン（Projection & Predicate Pushdown）： 50列あるテーブルから3列だけをクエリする場合、PyArrowはディスクからその3列に該当するバイト範囲のみを読み取ります。残りの47列は完全にスキップされます。CSVの場合、パーサーはファイル全体の先頭バイトから末尾バイトまでスキャンせざるを得ません。
  • ゼロコピー読み込み（Zero-copy read）： PyArrowは、C++のArrow C Data InterfaceからPythonランタイムへメモリマップファイル（Memory-mapped file）を直接マッピングします。この処理により、シリアライズやメモリの再割り当てに伴うオーバーヘッドが排除され、読み込み速度が劇的に向上します。

本番環境で6か月間安定運用を続けた結果、ジョブの処理時間は18分から42秒へと短縮されました。RAMのピーク値も以前のように16GBの上限に達することなく、780MB前後で安定して推移しています。以下に、具体的な実装アーキテクチャを解説します。

環境構築

PyArrowは、Linux、macOS、Windows向けにPyPI上でビルド済みのバイナリホイールが提供されています。最高レベルの圧縮を利用するために、追加で zstandard ライブラリもインストールしておきます。

# PyArrowと圧縮ライブラリzstandardのインストール
pip install pyarrow zstandard

# オプション: DataFrameのエクスポートなどで下位互換性が必要な場合にpandasをインストール
pip install pandas

インストール完了後、Pythonターミナルで圧縮コーデックが利用可能か素早く確認します。

import pyarrow as pa
import pyarrow.parquet as pq

print(f"PyArrow version: {pa.__version__}")
print(f"ZSTD supported: {pa.Codec.is_available('zstd')}")
print(f"Snappy supported: {pa.Codec.is_available('snappy')}")

両方の確認結果が True を返せば、高圧縮なParquetファイルを処理する準備は完了です。

本番環境での標準実装：読み込み、書き込み、最適化

1. 明示的なスキーマ定義によるParquet書き込み

Parquetファイルを書き込む際によくある落とし穴が、処理エンジンによる自動型推論（Type inference）に依存してしまうことです。これはCPUリソースを無駄に消費するだけでなく、異なるログバッチ間でスキーマの不一致（Schema mismatch）を引き起こす原因になります。最初から pa.schema を使って厳密に定義することを徹底しましょう。

import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime

# 1. スキーマを明示的に定義し、可能な限り最小のデータ型を選択
schema = pa.schema([
    ('event_id', pa.string()),
    ('user_id', pa.int64()),
    ('action', pa.string()),
    ('latency_ms', pa.float32()),  # float64の代わりにfloat32を使用してRAM消費を50%削減
    ('created_at', pa.timestamp('ms')),
    ('is_bot', pa.bool_())
])

# 2. サンプル配列データ
data = [
    pa.array(['evt_101', 'evt_102', 'evt_103']),
    pa.array([10001, 10002, 10003]),
    pa.array(['login', 'view_item', 'checkout']),
    pa.array([45.2, 120.8, 310.5], type=pa.float32()),
    pa.array([datetime.now(), datetime.now(), datetime.now()], type=pa.timestamp('ms')),
    pa.array([False, False, True])
]

# 3. PyArrow Tableの作成
table = pa.Table.from_arrays(data, schema=schema)

# 4. ZSTDレベル3および行グループサイズ50,000行でディスクに書き込み
pq.write_table(
    table,
    'events_2026_10.parquet',
    compression='zstd',
    compression_level=3,
    row_group_size=50000
)
print("Parquetファイルの書き込みに成功しました！")

ここで row_group_size パラメータの指定は極めて重要です。小さすぎる値（5,000行未満など）を設定するとメタデータが肥大化し、読み込み速度が目に見えて低下します。逆に1つの行グループに数百万行を詰め込むと、並列読み込みやパーティションのプルーニング性能が損なわれます。多くのバックエンドログ処理システムでは、50,000〜100,000行前後が最適なスイートスポットとなります。

2. 読み込みの最適化：プロジェクションとPyArrow Compute

数個のメトリクスを算出するだけなら、ファイル全体を読み込んではいけません。columns パラメータを使用して、必要なデータ列のみを取得します。

import pyarrow.parquet as pq
import pyarrow.compute as pc

# 必要な2列（user_idとlatency_ms）のみをメモリにロード
table_subset = pq.read_table(
    'events_2026_10.parquet',
    columns=['user_id', 'latency_ms']
)

print(f"レコード数: {table_subset.num_rows}")
print(f"使用RAM: {table_subset.nbytes / (1024 * 1024):.2f} MB")

# pandasに変換せず、C++ベースのPyArrow Computeで直接計算
avg_latency = pc.mean(table_subset['latency_ms'])
print(f"平均レイテンシ: {avg_latency.as_py():.2f} ms")

ポイントとして、pyarrow.compute による直接計算はメモリ上のC++データ構造をそのまま維持します。NumPyやpandasへの不要なデータコピー処理を回避できるため、バッチジョブのCPU負荷を大幅に削減できます。

3. データパーティショニング（Hive Partitioning）

データが数百万件から数億件規模に膨らむと、すべてを単一ファイルにまとめる設計は深刻なI/Oボトルネックを招きます。action=login/ や date=2026-10-27/ のように、ディレクトリ階層を用いたデータ分割を行いましょう。

import pyarrow.dataset as ds
import pyarrow.parquet as pq

# action列に基づいてディレクトリを自動分割してデータセットを書き込み
pq.write_to_dataset(
    table,
    root_path='events_partitioned',
    partition_cols=['action'],
    compression='zstd'
)

# Hiveパーティショニングを適用してDataset API経由でデータを読み込み
dataset = ds.dataset('events_partitioned', format='parquet', partitioning='hive')

# フィルタリング: PyArrowはaction=checkoutのフォルダのみを開き、他のフォルダは完全にスキップ
checkout_filter = ds.field('action') == 'checkout'
filtered_table = dataset.to_table(filter=checkout_filter)

print(f"スキャンされたcheckoutイベント数: {filtered_table.num_rows}")

ベンチマークとパフォーマンスモニタリング

パフォーマンスを感覚で推測してはいけません。以下のスクリプトを使えば、ディスク消費量とメタデータの読み出し時間を直接計測できます。

import os
import time
import pyarrow.parquet as pq

file_path = 'events_2026_10.parquet'

# 1. ディスク上の物理ファイルサイズ
size_mb = os.path.getsize(file_path) / (1024 * 1024)
print(f"[Disk Usage] ファイルサイズ: {size_mb:.2f} MB")

# 2. 実データをすべてロードせずにメタデータのみを読み込み
parquet_file = pq.ParquetFile(file_path)
meta = parquet_file.metadata
print(f"[Metadata] 総行数: {meta.num_rows:,}")
print(f"[Metadata] Row Group数: {meta.num_row_groups}")
print(f"[Metadata] 圧縮アルゴリズム: {meta.row_group(0).column(0).compression}")

# 3. 特定の1列の読み込み時間をベンチマーク
t0 = time.perf_counter()
col_data = parquet_file.read(columns=['user_id'])
elapsed = (time.perf_counter() - t0) * 1000

print(f"[Benchmark] 'user_id' 列の読み込み時間: {elapsed:.2f} ms")
print(f"[Memory] ロードされた列のRAM: {col_data.nbytes / (1024 * 1024):.4f} MB")

本番環境へデプロイする際は、次の3つの指標を監視することが重要です。

  1. 圧縮率（Compression Ratio）： メモリ上の生データサイズ（table.nbytes）とディスク上のファイルサイズを比較します。一般的なシステムログの場合、ZSTDレベル3で1:4〜1:7の圧縮比率を達成するのが目安です。
  2. スモールファイル問題（Small Files Problem）： ファイルサイズが50MB程度であるにもかかわらず、Row Groupが500個以上存在している状態はアンチパターンです。ヘッダーメタデータの読み込みオーバーヘッドにより、Parquetの強みが相殺されてしまいます。この傾向が見られる場合は、ディスクへのコミット前に小さなバッチを結合（Compact）してください。
  3. メモリピーク値（RSS）： Linux環境では /usr/bin/time -v python script.py コマンドを実行し、Maximum resident set size の項目を確認します。データレイク全体のデータ量増加に伴って線形に増大するのではなく、日次バッチで一定のメモリ使用量を維持できているかをチェックします。

CSVからParquetへの移行とPyArrowの組み合わせは、I/Oボトルネックの根本的な解消につながります。中規模から大規模なデータ処理パイプラインにおいて、サーバー費用の増額に踏み切る前に検討すべき、極めて効果的なリソース最適化アプローチです。

Share:
Tags: