以前pydanticでのデータ型検証やPolarsでのデータパイプライン応用を扱いましたが、pandasやPolarsのデータフレームそのものに対して「想定した形になっているか」を検証したい場面もあります。今回は、データフレーム専用のバリデーションライブラリ**Pandera**を整理します。
なぜデータフレーム専用の検証ライブラリが必要か
(1) 以前扱ったpydanticは、1件のレコード（辞書やオブジェクト）に対する型検証には強力ですが、数百万行のデータフレーム全体に対して「amount列は必ず0以上」「customer_id列に欠損値がない」といった検証をかけるのには不向きです。Panderaは、こうした「列全体・データフレーム全体に対する制約」を宣言的に表現できます。
(2) 以前扱ったGlue ETLやPolarsでのデータパイプラインでは、外部から取り込んだCSVやAPIレスポンスのデータ品質が保証されているとは限りません。パイプラインの入り口でPanderaによる検証を挟んでおくことで、想定外のデータが後続処理に流れ込むのを未然に防げます。
環境構築と基本スキーマ
uv add pandera
import pandas as pd
import pandera as pa
from pandera import Column, Check
schema = pa.DataFrameSchema({
"order_id": Column(int, Check.greater_than(0)),
"amount": Column(int, Check.greater_than_or_equal_to(0)),
"status": Column(str, Check.isin(["pending", "shipped", "delivered"])),
})
df = pd.read_csv("orders.csv")
validated_df = schema.validate(df)
(1) `DataFrameSchema`に各列の型と制約（`Check`）を定義し、`validate`を呼ぶだけでデータフレーム全体を検証できます。制約に違反する行があれば、以前扱ったpydanticのバリデーションエラーと同様、`SchemaError`が送出され、どの列・どの値が問題だったかが具体的に示されます。
クラスベースでのスキーマ定義
import pandera as pa
from pandera.typing import Series
class OrderSchema(pa.DataFrameModel):
order_id: Series[int] = pa.Field(gt=0)
amount: Series[int] = pa.Field(ge=0)
status: Series[str] = pa.Field(isin=["pending", "shipped", "delivered"])
class Config:
strict = True
@pa.check_types
def load_orders(path: str) -> pa.typing.DataFrame[OrderSchema]:
return pd.read_csv(path)
(1) この`DataFrameModel`によるクラスベースの書き方は、以前扱ったpydanticの`BaseModel`とほぼ同じ見た目で定義できます。`@pa.check_types`デコレータを関数につけておくと、戻り値のデータフレームが自動的に検証され、以前扱ったデコレータによる共通化の考え方がここでも活きています。
(2) `strict = True`を指定すると、スキーマに定義されていない列が含まれている場合もエラーになります。想定外の列が紛れ込んでいないかまで含めて検証したい場合に有効です。
列間の関係を検証するカスタムチェック
class OrderSchema(pa.DataFrameModel):
order_id: Series[int] = pa.Field(gt=0)
subtotal: Series[int] = pa.Field(ge=0)
tax: Series[int] = pa.Field(ge=0)
total: Series[int] = pa.Field(ge=0)
@pa.dataframe_check
def total_matches_subtotal_plus_tax(cls, df: pd.DataFrame) -> Series[bool]:
return df["total"] == df["subtotal"] + df["tax"]
(1) 単一の列だけでなく、複数の列を組み合わせた業務ルール（この例では「合計金額が小計と税額の和と一致するか」）も`dataframe_check`で表現できます。以前扱ったRDSでの制約（CHECK制約）に近い発想を、DBに保存する前のPython側で先に確認できるイメージです。
Polarsとの連携
import polars as pl
import pandera.polars as pa
class OrderSchema(pa.DataFrameModel):
order_id: int = pa.Field(gt=0)
amount: int = pa.Field(ge=0)
df = pl.read_csv("orders.csv")
validated_df = OrderSchema.validate(df)
(1) 以前扱ったPolarsでのデータパイプラインに対しても、`pandera.polars`モジュールを使えば同様のスキーマ検証が行えます。pandas版とほぼ同じ書き方で移行できるため、データフレームライブラリを切り替えた場合でも検証ロジックを大きく書き直す必要がありません。
ETLパイプラインへの組み込み
def transform_orders(raw_df: pd.DataFrame) -> pd.DataFrame:
validated_df = OrderSchema.validate(raw_df, lazy=True)
return validated_df.assign(amount_with_tax=lambda d: d["amount"] * 1.1)
(1) `lazy=True`を指定すると、最初のエラーで処理を止めるのではなく、すべての検証エラーをまとめて収集してから報告してくれます。以前扱ったGlue ETLのような大規模なバッチ処理では、1件ずつエラーを直しては再実行するより、まとめて問題点を洗い出せる方が効率的です。
テストコードとの組み合わせ
import pytest
def test_load_orders_schema():
df = load_orders("tests/fixtures/sample_orders.csv")
OrderSchema.validate(df)
(1) 以前扱ったpytestのテストコードにスキーマ検証を組み込んでおくと、外部から取り込むサンプルデータの形式が変わった場合に、CIの段階で気づけるようになります。以前扱ったGitHub ActionsのCIパイプラインに組み込めば、データ品質のチェックもテストの一部として自動化できます。
まとめ
Panderaは、データフレーム全体に対して型・値の範囲・列間の関係といった制約を宣言的に検証できるライブラリです。以前扱ったpydanticが1件のレコードの検証に向いていたのに対し、Panderaは大量の行を持つデータフレームの検証に向いており、両者は対象とするデータの粒度が異なります。ETLパイプラインの入り口や、テストコードの一部として組み込んでおくことで、データ品質の劣化に早く気づける体制を作れます。
