はじめに — つまずきに寄り添って
大きなCSVを扱うと、気づかないうちにメモリを使い切り、ジョブが途中で落ちることがあります。まずは「自分の環境でどこまで一括読み込みできるか」を把握することが重要です。本記事では、現場で実際に使える手順とコードパターンを示します。すぐ試せるチェックリストと再開性の工夫も含め、落ち着いて運用できる形で説明します。
イントロ：なぜ一括読み込みが危険か／いつストリーミング処理を選ぶか
一括読み込みはコードがシンプルになりますが、RAMを超えるとプロセスが強制終了します。判断の目安を簡潔にまとめます（目安は経験値であり、データのカラム数・型で変動します）。
| 条件 | 目安 | 推奨アクション |
|---|---|---|
| ファイルサイズ | 数GB以上（特に10GB超） | ストリーミング／チャンク処理 |
| 行数 | 数百万行以上 | チャンク分割＋列最適化 |
| 作業環境のRAM | 利用可能RAMがデータの1.5倍未満 | ストリーミングを選ぶ |
ジェネレータとイテレータの実務パターン
ジェネレータはメモリを使わずに行単位で処理できます。withと組み合わせるとファイルハンドルの解放が確実です。
CSVを行単位でストリーミングする簡単な例：
実装メモ: コード例は環境に合わせて調整してください。例: import csv
ポイント：
- ヘッダを別で取得することで、行ごとの処理コードはヘッダに依存しない。
- 例外は呼び出し側でハンドルして、必要ならチェックポイント（行番号）を保存する。
pandasのチャンク処理
pandasのread_csv(chunksize=…)は既存のDataFrame APIを使いながら分割処理できます。1チャンクあたりのサイズは行数ではなくメモリ使用量を意識して決めます。
実装メモ: コード例は環境に合わせて調整してください。例: import pandas as pd
注意点：
- parse_datesは便利だが遅い。必要なカラムだけ指定する。
- dtypeを明示しておくとメモリと速度が安定する。
- チャンク内でスキーマ変換すると全体の整合性が崩れないか確認する（第118回のスキーマ検証参照）。
メモリ最適化の具体手法
一般的な手法を表にまとめます。実務では複数を組み合わせます。
| 手法 | 効果 | 実装のヒント |
|---|---|---|
| 数値のダウンキャスト | メモリ削減（float64→float32など） | pandasのastypeやto_numericのdowncast引数を活用 |
| カテゴリ型化 | 離散値のメモリ削減（文字列列など） | pd.Categoricalまたはastype(‘category’) |
| 不要列削除 | 即時効果 | 読み込み時にusecolsで限定する |
| object列の扱い | 大量のユニーク文字列はメモリを圧迫 | 必要ならハッシュ化や部分列保持 |
| 計測 | どこでボトルネックか把握 | psutilやmemory_profilerでプロファイル |
簡単なダウンキャスト例：
実装メモ: コード例は環境に合わせて調整してください。例: def downcast_df(df):
列指向フォーマットへの変換：Parquet / Feather
ParquetやFeatherは列指向で圧縮効率が高く、読み戻しも高速です。業務での使い分け方：
| フォーマット | 利点 | 注意点 |
|---|---|---|
| Parquet | 高圧縮・スキーマ保存・Spark互換 | 小さなファイルが多数になると管理が面倒 → partitioningで対応 |
| Feather | 高速な読み書き（メモリマップ） | 圧縮オプションが限定的、フォーマット差に注意 |
分割戦略例：日付や顧客IDのハッシュでpartitioningし、処理単位で読み込むと効率的です。
バッチ推論ワークフロー
チャンク→バッチ化→モデルAPI呼び出しの一般的な流れを示します。APIのレイテンシや制限を踏まえ、同期・並列・非同期を使い分けます。
同期で並列化する例（concurrent.futuresを使う）：
実装メモ: コード例は環境に合わせて調整してください。例: from concurrent.futures import ThreadPoolExecutor, as_completed
非同期（asyncio）は高レイテンシAPIや大量の小さなリクエストに向きます。どちらを選ぶかはI/O待ち時間とCPU処理量で判断します。
レート制限と指数バックオフの例（擬似コード）：
実装メモ: コード例は環境に合わせて調整してください。例: def with_backoff(callable, max_retries=5):
障害対策と再開性
途中で失敗しても再開できる設計が重要です。よく使うパターン：
- チェックポイント：処理した最大行IDやファイルオフセットを定期的に保存する（JSONやDBに保存）。
- 部分出力のマージ：チャンクごとに別ファイルに出力し、最後にマージ。失敗時は未処理チャンクだけ再実行。
- 冪等性（idempotency）：同じレコードを複数回処理しても結果が壊れないAPI設計や、リクエストにidを付ける。
チェックポイント例（JSON保存の最小例）：
実装メモ: コード例は環境に合わせて調整してください。例: checkpoint = {‘file’: ‘large.csv’, ‘last_row’: 123456}
エンドツーエンドの手順テンプレート
実務で使える簡単なテンプレート（順序）：
- 1. 環境確認：利用可能RAM、ディスク、APIレート制限を把握
- 2. スモークテスト：サンプル（1万行）でチャンク処理→推論→保存を試す
- 3. 読み込み：ジェネレータまたはpd.read_csv(chunksize)
- 4. 変換：型指定・ダウンキャスト・カテゴリ化
- 5. スキーマ検証：第118回の手法で検証
- 6. バッチ推論：バッチ化＋並列／非同期でAPI呼び出し
- 7. 後処理と永続化：ParquetやDBへ保存
- 8. 監視：処理時間・失敗率・APIレイテンシを記録
簡易コードスケルトン：
実装メモ: コード例は環境に合わせて調整してください。例: for chunk in pd.read_csv(‘large.csv’, chunksize=100_000):
実務の落とし穴と運用チェックリスト
| 項目 | 確認方法 | 即時の対処 |
|---|---|---|
| メモリ不足 | psutilでRAM推移を監視 | チャンクサイズ削減／列削除 |
| APIタイムアウト・レート超過 | 失敗率・429応答をログ | バックオフ・バッチサイズ調整 |
| スキーマ不一致 | スモークテストで差分確認 | スキーマ変換ルールを追加 |
| 部分データの重複 | 出力の重複チェック | 冪等化ロジックを導入 |
運用で常に見ると良いメトリクス：処理時間（チャンクごと）、成功率、API平均レイテンシ、メモリ使用率。
まとめ
本記事では、大容量CSV／表データを現場で安定して処理するための実務パターンを紹介しました。ポイントは以下です。
- ファイルサイズや利用可能RAMに応じて、最初からストリーミングを選ぶ判断をすること。
- ジェネレータ／pandasチャンク処理でメモリを節約しつつ、型指定やダウンキャストでさらに削減すること。
- Parquet等の列指向フォーマットに変換すると運用が楽になるが、partition戦略を設計すること。
- バッチ推論はチャンク→バッチ→API呼び出しの流れ。並列／非同期とバックオフを組み合わせること。
- チェックポイントや部分出力の保存で再開性を確保し、冪等化を意識すること。
次の一歩（提案）：
- 実験課題：1万行のCSVでチャンク処理→小さなLLMバッチ推論→Parquet保存を試す
- 続きの候補記事：ストリーミングETLの監視化、S3やデータレイクとの連携方法
読者の方がまずやるべきは、小さなスモークテストを作って運用フローを確かめることです。疑問や具体的な環境（RAM量や処理時間の目安）があれば、そこに合わせた細かい調整案を提示しますのでお知らせください。