はじめに — つまずきに寄り添う一言
CSVやJSONLを扱うと、欠損や型、行の重複、結合ルールといった「地味に面倒な問題」に時間を取られがちです。本記事では、現場で最も出番の多い「リスト（行の列挙）と辞書（行を表す）」を使った実務的な処理パターンを、すぐ使える関数設計と運用チェックリスト付きで整理します。134/135回のファイル入出力・クリーニング回とつなげて読めるよう設計しています。
基本方針（入出力分離・小さな再利用可能コンポーネント）
関数はできるだけ「入力（rows）を受け取って新しい出力を返す」スタイルにします。ミュータブルな引数を書き換えるとバグになりやすいため、状態変更は明示的に行い、副作用は最小限にします。型ヒントと短いdocstringをつけると再利用性が上がります。
よく使う基本パターン（一覧と例）
| パターン | 関数署名（例） | 説明・使い方（例スニペット） |
|---|---|---|
| フィルタ | def filter_rows(rows: list, pred) -> list | rows = [r for r in rows if pred(r)] 例: pred = lambda r: int(r[‘qty’]) > 0 |
| 変換（map） | def transform_rows(rows: list, fn) -> list | rows = [fn(r.copy()) for r in rows] 例: fnで日付正規化や型変換 |
| グルーピング | def group_by(rows: list, key_fn) -> dict | collections.defaultdict(list)で集約。agg関数で合計や平均を作る |
| マージ（アップサート） | def merge_rows(base, updates, key: str, prefer=’updates’) -> list | 主キーでインデックス化し優先度ルールで列の上書きと衝突ログを出す |
| チャンク処理（大規模データ） | def chunked_iter(it, size: int) | yieldでメモリを節約。itertools.isliceで実装 |
集約とグルーピングの使い分け
| 手法 | 適用例 | メモリ | ソート要否 |
|---|---|---|---|
| collections.defaultdict(list) | 任意のキーでまとめて集約・集計 | キー数×行参照分のメモリ | 不要 |
| collections.Counter | 出現頻度や簡単な頻度集計 | キー数分のメモリ | 不要 |
| itertools.groupby | ソート済みデータの連続ブロック処理（ストリーミング向け） | ほぼ定数（ソート前提） | 必要（事前に同キーでソート） |
軽い比較ベンチマークのポイント: defaultdictはランダム順のまま使える反面、キー数が多いとメモリが増える。groupbyは事前ソートが必要でソートコストが主要なボトルネックになります。
マージ（アップサート）パターンの実務設計
運用的に重要なのは「冪等性」と「衝突の可視化」です。典型的な実装手順は次のとおりです。
- baseをkey→rowの辞書に変換する（index化）
- updatesを走査して既存行があれば優先度ルールで列を更新、なければ挿入
- 変更点はログ（衝突レコード）として出力する
- 最後に辞書をリスト化して返す（順序が必要なら元順序を保持）
| ポイント | 注意点 |
|---|---|
| 優先度設計 | 列ごとに“どちらを優先するか”を明示するルール表を運用ドキュメント化する |
| 冪等性 | 同じ更新を何度適用しても結果が変わらないように設計する（タイムスタンプ比較など） |
| 衝突ログ | 変更前後の値をCSV/JSONで出力して監査できるようにする |
大規模データ対策（チャンクとストリーミング）
| 技法 | 利点 | 実務向け注意点 |
|---|---|---|
| ジェネレータ + itertools.islice | メモリ使用量を一定に抑えられる | ステートフル処理（グルーピングなど）はチャンク境界の扱いに注意 |
| 外部ソート（重複除去） | メモリに乗らないキーでのユニーク化が可能 | ディスクI/Oのコストを見積もること |
| ハッシュリング（簡易） | キーをハッシュで分割し、分割ごとに処理・マージする | 分割数設計とファイル管理が運用コストになる |
小さな再利用コンポーネント設計の例
| 関数 | 署名 | 戻り値 | docstringのポイント |
|---|---|---|---|
| filter_rows | filter_rows(rows: list, pred: Callable[[dict], bool]) -> list | フィルタ後の行リスト | predの期待するキーと型、欠損時の動作を明記 |
| merge_rows | merge_rows(base, updates, key: str, prefer: str=’updates’) -> list | マージ後の行リストとオプションで衝突ログ | 優先ルールと冪等性の挙動を説明 |
pytestでのテスト例（境界ケース・欠損値）
| テスト名 | 簡単な内容（擬似アサーション） |
|---|---|
| test_filter_missing_key | assert filter_rows([{‘a’:’1′},{ }], lambda r: r.get(‘a’)==’1′) == [{‘a’:’1′}] |
| test_merge_prefer_updates | assert merge_rows([{‘id’:’1′,’v’:’old’}],[{‘id’:’1′,’v’:’new’}],’id’,prefer=’updates’)[0][‘v’]==’new’ |
| test_chunked_iter_boundary | assert len(list(chunked_iter(range(5), 2))) == 3 |
実務運用チェックリスト
| チェック項目 | 目的・方法 |
|---|---|
| エンコーディング確認 | ファイル読み込み時にencodingを明示（UTF-8/BOMなど） |
| 欠損値ルール | 欠損の扱い（除外/補完/デフォルト値）を自動チェックで実行 |
| 型変換の検証 | サンプル変換後の型チェックと失敗行の報告 |
| 行数差分とサンプルハッシュ | 入出力の行数、先頭N行のハッシュで大きなズレを検出 |
| 監査ログ出力 | マージや上書きの前後をログ化（143回の監査ログ方針に準拠） |
落とし穴と改善策（実務でよくある問題）
- ミュータブルな行をそのまま編集して意図せず元データを書き換える: コピーして操作する。
- キーの正規化ミス（空白・大文字小文字）: 正規化ルールを統一して統合テストに入れる。
- パフォーマンスの盲点: プロファイラでボトルネックを特定し、必要ならpandas導入を判断（全体の読みやすさと運用コストを比較）。
次の一歩（ワークフロー化への橋渡し）
作成したコンポーネント群をPrefectやcronに組み込み、リトライ、監査ログ、テストを回すことで運用に耐えるフローになります。まずは1つの取り込み->正規化->マージの小さなパイプラインを作り、SLO（処理時間・成功率）を設定してください。
まとめ
リストと辞書は、表データ処理の基本であり続けます。重要なのは「小さな、再利用可能で副作用の少ない関数」を設計すること、そして運用チェックリストで品質を保つことです。本記事のパターンをテンプレートにして、まずは小さなパイプライン一つを作ってみてください。次回以降はワークフロー化と監査自動化の具体例へつなげます。