はじめに — データ受け口でつまずいていませんか
CSVやAPIから受け取った表データをそのままモデルや自動処理に流すと、型エラーや欠損、想定外の値で処理が止まります。実務では「どの検証をどこで」「どのくらい厳しく」実施するかを決め、現場で回すことが重要です。本記事では、まずその日に試せる最小実装（pandas + ゼロ依存のバリデータ）を示し、導入→ローカル検証→CI→運用監視までの実務的手順を解説します。
データ検証で優先すべきルール
まずは優先度の高い検証項目を整理します。下の表は各ルールの目的と現場での取り扱い方の要点です。
| 検証ルール | 目的 | 現場の扱い（実務上の判断） |
|---|---|---|
| 型（型変換） | 処理前提のデータ型を担保する | まずは厳格に検出→自動補正はログ必須。補正の閾値を運用で管理。 |
| 必須（存在チェック） | 処理に必須の列や値が欠けていないか | 欠損は明確にエスカレーション。許容する場合は補完方針をSOP化。 |
| 範囲／フォーマット | 想定外の外れ値や形式不一致を検出 | 閾値違反はサンプリングしてヒューマンチェック。閾値は更新履歴を残す。 |
| 一意性 | キー重複による上書きや二重処理を防ぐ | 重複は原則エラー。バッチ単位で差分チェックを行う。 |
| 欠損の扱い | 削除・補完・エスカレーションの判断基準を明確に | 削除する場合は影響範囲を事前評価。補完は別列で補完理由を出力。 |
最小実装ハンズオン：pandasで素早く検証する
ここでは「依存を小さく」保った実装例を示します。前提として pandas が利用できる環境を想定します（pip install pandas）。
1) スキーマ定義（辞書形式）
スキーマは簡潔な辞書で定義します。業務ごとにこの辞書を更新します。
schema = {
'id': {'type': 'int', 'required': True, 'unique': True},
'name': {'type': 'str', 'required': True, 'unique': False},
'age': {'type': 'int', 'required': False, 'min': 0, 'max': 120},
'score': {'type': 'float', 'required': True, 'min': 0.0, 'max': 100.0},
'joined': {'type': 'date', 'required': True, 'format': '%Y-%m-%d'}
}
2) 安全な read_csv（例）
まずは全列を文字列で読み、後で明示的に変換します。これにより想定外の変換で失敗するリスクを減らせます。
import pandas as pd
def safe_read_csv(path):
return pd.read_csv(path, dtype=str, keep_default_na=False)
3) 列単位／行単位のバリデータ（概念実装）
主要なチェック関数を示します。実務ではログ出力やエラーファイル出力を組み合わせます。
from datetime import datetime
def convert_type(series, spec):
t = spec.get('type')
if t == 'int':
return pd.to_numeric(series, errors='coerce').astype('Int64')
if t == 'float':
return pd.to_numeric(series, errors='coerce')
if t == 'date':
fmt = spec.get('format')
return pd.to_datetime(series, format=fmt, errors='coerce')
return series.astype('string')
def validate_dataframe(df, schema):
errs = []
df2 = df.copy()
# 型変換
for col, spec in schema.items():
if col in df2.columns:
df2[col] = convert_type(df2[col], spec)
else:
if spec.get('required'):
errs.append({'row': None, 'col': col, 'error': 'missing_column'})
# 列単位チェック（範囲・必須）
for col, spec in schema.items():
if col not in df2.columns:
continue
s = df2[col]
# 必須
if spec.get('required'):
missing_idx = s.isna() | (s == '')
for i in df2[missing_idx].index.tolist():
errs.append({'row': int(i), 'col': col, 'error': 'required_missing'})
# 範囲
if spec.get('type') in ('int', 'float'):
if 'min' in spec:
bad = s[s < spec['min']]
for i in bad.index.tolist():
errs.append({'row': int(i), 'col': col, 'error': 'below_min'})
if 'max' in spec:
bad = s[s > spec['max']]
for i in bad.index.tolist():
errs.append({'row': int(i), 'col': col, 'error': 'above_max'})
# 一意性チェック
for col, spec in schema.items():
if spec.get('unique') and col in df2.columns:
dup = df2[df2.duplicated(subset=[col], keep=False)][col]
for i in dup.index.tolist():
errs.append({'row': int(i), 'col': col, 'error': 'not_unique'})
return df2, errs
4) 失敗時のサンプル出力とエラーファイル
検出したエラーはCSVに出力し、オペレーターが原因を追跡できるようにします。
def dump_errors(df, errs, out_path='errors.csv'):
rows = []
for e in errs:
r = {'row': e['row'], 'col': e['col'], 'error': e['error']}
if e['row'] is not None:
r['value'] = df.iloc[e['row']].get(e['col'])
rows.append(r)
import csv
keys = ['row', 'col', 'error', 'value']
with open(out_path, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=keys)
writer.writeheader()
writer.writerows(rows)
5) 簡易CLI例
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('input')
parser.add_argument('--errors', default='errors.csv')
args = parser.parse_args()
df = safe_read_csv(args.input)
df2, errs = validate_dataframe(df, schema)
if errs:
dump_errors(df, errs, args.errors)
print(f'Validation failed: {len(errs)} issues. See {args.errors}')
raise SystemExit(1)
else:
print('Validation passed')
# 次の処理へ渡す（例: df2.to_csv('clean.csv', index=False))
6) pytest を使ったユニットテストの例
def test_missing_required(tmp_path):
import pandas as pd
df = pd.DataFrame({'id': ['1'], 'name': ['']})
_, errs = validate_dataframe(df, schema)
assert any(e['error'] == 'required_missing' for e in errs)
拡張編：既存スキーマライブラリとの比較と使い分け
プロトタイプはゼロ依存で速く回せますが、規模が大きくなると既存ライブラリの導入を検討します。下表は現場での使い分けの目安です。
| 目的 | ゼロ依存（今回の実装） | pandera / pydantic / Great Expectations |
|---|---|---|
| 素早いプロトタイプ | 最適 — 依存少なく即導入可 | 導入コストあり |
| 複雑な型変換・再利用可能なスキーマ | コードが膨らむ | 有利（明示的・テストしやすい） |
| レポート／ドキュメント出力・データプロファイリング | 自作が必要 | Ready-made 機能あり（Great Expectations 等） |
| 運用の堅牢性 | 簡潔だが手作業が増える | 堅牢なフレームワークがある |
運用編：ログ・アラート・CI・ロールバック
検証は導入後も継続的に監視する必要があります。以下は実務で押さえるべきポイントです。
- ログ出力：バリデーション結果は構造化ログ（JSON）で残す。行数やエラー種別をメトリクス化する。
- アラート：エラー率が閾値（例：パイプライン処理件数に対して5%）を超えたら通知。閾値は履歴でチューニング。
- CI：新しいスキーマや変換ロジックはユニットテストと統合テストを用意。GitHub Actions で csv サンプルを検証するワークフローを自動化する。
- ロールバック手順：自動処理で不正データが流れた場合、原則は旧データでの再実行とログによる差分復元手順をSOPに記載。
- サンプリング戦略：フル検証コストが高い場合、ランダムサンプリングと重み付きサンプリングを組み合わせて監視。
チェックリストと現場での落とし穴
導入前後に確認すべきチェックリストを示します。短い表で優先順位を付けています。
| 項目 | 必須度 | コメント |
|---|---|---|
| スキーマのバージョン管理 | 高 | 変更履歴を明記し、互換性ルールを定義する。 |
| エラーファイルの保管期間 | 中 | 原因追跡のため一定期間は保存。 |
| 自動補正のログ | 高 | 補正が行われた場合は理由と原値を保存。 |
| アラート閾値の設定 | 高 | 運用開始後に経験値でチューニング。 |
| SOP（標準作業手順書）への落とし込み | 高 | 誰が何をいつまでに行うかを明確にする。 |
簡単な運用フロー（要点）
- 受信→safe_read_csvで読み込み→validate_dataframeで検証→問題があればerrors.csv出力・アラート→問題なければ次処理へ
- CIでサンプルデータとスキーマを常時検証、スキーマ変更はPRで承認するフローを必須化
- 重大なエラーは手動対応ログを残し、再発防止策をSOPに追加
まとめ
本記事では、まずはその日中に試せる「pandasを使った最小実装」を提示しました。実務では単にバリデータを作るだけでなく、スキーマのバージョン管理、ログとエラーファイル、CI による自動検証、アラート設計、SOP への落とし込みが重要です。初期はゼロ依存の実装で素早く回し、業務が拡大したら pandera や Great Expectations のようなフレームワーク導入を検討すると良いでしょう。
到達目標：この記事を読んだら、まずは safe_read_csv・schema 辞書・validate_dataframe を使ってサンプルCSVを検証し、errors.csv を出力する最小実装を作成してください。その上で、ユニットテストを追加し、CI に組み込む流れを試してください。
次回は第113回・第115回で触れたCSV入出力と変換の実践例を踏まえ、実際のパイプラインに組み込むテンプレートを紹介します。