日々の業務で「CSVを読み込んで整形したい」「行ごとの欠損や型変換でつまずく」「大量データでメモリが足りない」と感じたことはありませんか。この記事では、現場でよくある「CSV→正規化→集計→モデル入力」の一連処理を、リスト・辞書・for/if を中心に段階的に示します。第113回（CSV読み書き）・第114回（関数とモジュール設計）の知見を活かし、今日中に試せるコードとチェックリストを提供します。
導入: 現場の課題とこの記事のゴール
現場でよくあるケースを想定します。
- 受注CSVに日付が文字列、数量が空文字やマイナスで混在している。
- 複数ファイルを正規化してから顧客単位で集計し、機械学習モデルの入力バッチを作る必要がある。
本記事のゴールは、再現可能で堅牢な変換パイプラインを作ることです。具体的には:
- 行単位の正規化関数を作る
- キーの正規化、フィルタ・マッピング、集約を段階的に実装する
- メモリに優しいバッチ化とエラー処理を加える
前提とセットアップ
前提環境:
- Python 3.8+（3.10を推奨）
- 推奨エディタ: VS Code / PyCharm
- 参考: 第113回での csv モジュール説明、第114回での関数分割の方針を踏襲
最小サンプルCSV（コピー&ペーストで試せます）:
実装メモ: コード例は環境に合わせて調整してください。例: order_id,customer_id,order_date,quantity,price
この記事のコードは、標準ライブラリのみで動くようにしています。必要に応じて pandas 等を導入してください（ただし小規模スクリプトは標準ライブラリで十分なことが多いです）。
基本パターン: リストと辞書の使い分け
行データの扱いは大きく二つのスタイルがあります。読みやすさと操作のしやすさで使い分けます。
| 形 | 長所 | 短所 | 実務での使いどころ |
|---|---|---|---|
| 行リスト（[‘1′,’1001′,…’]） | 軽量、順序保持 | 列名参照が面倒 | 高速に単純処理するバッチ |
| レコード辞書（{‘order_id’:’1′,…}） | 列名で参照でき可読性高い | メモリ増（キー情報） | 正規化・集約・検証処理 |
読み込み例（csv.DictReader を使うと辞書が得られます）:
実装メモ: コード例は環境に合わせて調整してください。例: import csv
リスト内包表記・辞書内包表記の使いどころ
- 短い変換なら内包表記で可読かつ短く書ける
- 複雑な検証やログが必要ならforループで段階的に処理する
変換処理のステップ実装
ここでは、段階的に関数を作り、組み合わせてパイプラインにします。まずサンプルデータを辞書リストとして読み込んだものと仮定します。
1) 行の正常化（型変換・日付パース・空値処理）
実装メモ: コード例は環境に合わせて調整してください。例: from datetime import datetime
ポイント: 個別の try/except で失敗行を部分的に扱い、後の段階でスキップやログを決めると柔軟です。
2) キー正規化（dict.get / setdefault / defaultdict）
複数ソースを統合するときにキー名が異なる場合があります。setdefault や collections.defaultdict が便利です。
実装メモ: コード例は環境に合わせて調整してください。例: from collections import defaultdict
3) フィルタとマッピング（map 相当）
不要な行を除外しつつ、必要なフィールドへ変換します。
実装メモ: コード例は環境に合わせて調整してください。例: def filter_valid(rows):
4) 集約（groupby や累積集計）
少量データなら辞書で集計、順序付き集約が要る場合は itertools.groupby を使います。
実装メモ: コード例は環境に合わせて調整してください。例: def aggregate_by_customer(rows):
バッチ化とチャンク処理（メモリ対策）
大量CSVでは一括読み込みは避け、ジェネレータ/チャンク処理を使います。
実装メモ: コード例は環境に合わせて調整してください。例: import csv
チャンクサイズの選び方の目安:
| 用途 | 推奨チャンクサイズ |
|---|---|
| API呼び出し（レート制限あり） | 小〜中（10〜100） |
| モデル推論（GPU利用） | 中〜大（32〜512、モデル入力に依存） |
| 単純集計 | 大（1000〜） |
堅牢化: バリデーション・例外処理・ログ
実務では失敗しても原因が追える設計が重要です。
- 入力バリデーション: 必須カラムの存在チェック
- try/except の粒度: 行単位で捕まえて処理を継続する
- ログ出力: 処理開始/終了、バッチごとの集計、スキップ行は理由を残す
- 再試行・スキップ: 外部サービス呼び出しは指数バックオフで再試行
| 状況 | 戦略 |
|---|---|
| 一時的なAPIエラー | 再試行（2〜3回）→ログ→次に進む |
| データ整合性エラー（必須カラム欠落） | スキップ＋監査ログへ記録 |
| 致命的なフォーマット破損 | 処理停止＋アラート |
パフォーマンスとメモリの注意点
実務でよくある落とし穴と簡単な診断法:
- 浅いコピー vs 深いコピー: 大きな辞書を不用意に copy するとメモリ増
- 参照のまま変更するか、明示的に新しいオブジェクトを作るかを設計で決める
- 簡易プロファイリング: timeit, cProfile でホットスポットを特定する
| 問題 | 対処法 |
|---|---|
| メモリ使用量が多い | ジェネレータ化・チャンク化・不要なコピーを削除 |
| 処理が遅い | 鍵アクセスの回数削減・数値演算をまとめる・必要なら numpy/pandas を検討 |
テストとドキュメント
変換ロジックは小さな関数に分けて単体テストを書きます。pytest のサンプル:
実装メモ: コード例は環境に合わせて調整してください。例: def test_normalize_row():
最小ドキュメントテンプレート（チーム共有用）:
| 項目 | 記載例 |
|---|---|
| 入力形式 | orders.csv: order_id,int; order_date,YYYY-MM-DD or YYYY/MM/DD |
| 出力仕様 | customer_summary.json: customer_id, total_amount(float), total_qty(int) |
| エラー処理 | 欠損はスキップ、ログに記録。致命的エラーはアラート。 |
実務チェックリストと次の一歩
デプロイ前に確認する監視ポイント:
| チェック項目 | 確認内容 |
|---|---|
| 再現性 | 同じ入力で同じ出力が得られるか（ランダム要素なし） |
| ログと監査 | スキップ/エラーの行がログに残るか |
| パフォーマンス | 処理時間とメモリ使用量が許容範囲内か |
| バックアップ | 入力ファイルのアーカイブ方針があるか |
次の一歩: パイプラインのオーケストレーション（Airflow 等）や第104回のオーケストレーション記事での運用化を検討してください。
まとめ
- リストは軽量処理、辞書は可読性重視。用途に応じて使い分ける。
- 正規化→キー整備→フィルタ→集約の順で関数化するとテストしやすい。
- 大容量はジェネレータ・チャンク処理でメモリを抑え、バッチサイズは用途に応じて調整する。
- 堅牢化（ログ・バリデーション・再試行）の設計が運用で効いてくる。
この記事で示した小さなパイプラインを基に、まずは自分のCSVで一度試してみてください。問題が出た箇所がそのまま改善点になります。Manage AI の次回記事では、パイプラインのオーケストレーションと運用監視について触れる予定です。