Panderaをマスターしよう（1. 基本編）

はじめに

この記事では業務でpandas.DataFrameとpanderaを扱ったため、他人にも理解でき、作業を効率化できるよう、紹介いたします。

データフレームはどのような列・値が存在していても一律でpd.DataFrameとして扱うことが多く、その中身はコードを隅々まで読まないとわからないブラックボックス状態になってしまいます。

• チームメンバーに処理を引き継ぐ場合、伝えることが難しい。
  dfをインプットとして用いるとき、大抵列名を指定してそのスライスを見ることが多いです。インプットにするならばどのような列があるのか、その値は相応しいものなのかがわからなくなります。チームメンバーから「このdfにはどのような列がありますか？」「必要な値はなんていう列の中に入っていますか？」とチームメンバーから質問されることも多くありました。
  
  

• しっかりとdfを構成しないと、後々のプロセスのエラーとなることがある。
  誤ったdfを後々のプロセスに渡しても、その時点ではプロセスにエラーが起こらず、後々まで気づかない場合があります。後のプロセスでエラーが発生した場合、前の処理に遡ってエラー修正を行わなければならないうえ、dfを用いるたびに「このdfに誤りはないか」と疑心暗鬼にならざるを得なくなります。
  
  

• ケース毎に、列の有無が異なる場合がある。
  dfの操作として、複数のdfをpd.concatで結合することが多々あります。列が異なるdfを結合するとき、dfが空（empty）になるようなケースとそうではないケースでは、結合後のdfの列は異なります。この時存在しない列を指定してスライスを取ろうとすれば、もちろんエラーが発生します。
  
  

この様な問題を解決するのが、panderaです。

panderaの基本

この章で、panderaの基本的な使い方を紹介します。これを

panderaとは

panderaとは、Union.aiが開発したオープンソースプロジェクトで、pandas・polars・pyspark.pandasなどのデータフレーム型のオブジェクトの構造やデータの値が期待通りかどうかを検証（バリデーション）を行うためのツールです。

panderaを使うメリットとして、以下のようなものがあります。

• スキーマを見れば、データフレームがどのような列を持っているのか明記できる
• DataFrameが期待された列や要素を持っているかの検証（バリデーション）が行える
• 値の型が違うなど、期待されるものと実際のdfが異なる場合、期待される形になるように修正する

特に、アサーションとデータフレームの各列についてのメモを同時にこなせるのが1番の強みだと思います。

panderaは他にも仮説検証など様々な機能がありますが、ここでは上記の機能を主に解説します。

panderaのインストール

panderaはpython標準ライブラリではないため、インストールする必要があります。
numpyやpandasのように、pipを用いてインストールが可能です。

pip install pandera

使い方要約

こんな形でpanderaはスキーマとバリデーションを行っているんだよ、という概略です。これさえ知っていればスキーマを自分から書けずとも、他の人が書いたスキーマがわかるようになると思います。
ここでは、このようなデータフレームの評価を行っていきます。

python

import pandas as pd

data = pd.DataFrame(
    {
        "名前": ["山田太郎", "鈴木花子", "佐藤次郎"],
        "年齢": [25, 30, -5],
        "メールアドレス": ["[e-mail]", "suzuki@example", "[e-mail]"],
        "購入金額": [1200.5, None, -500.0],
    }
)

以降、import pandas as pdを前提として紹介していきます。

1. ファイル内でpanderaをimportする。

インストールできたら、ファイル上でimportします。pandasがpdと略されるように、panderaはpaと略されることが多いようです。

python

import pandera as pa

2. スキーマを定義する。

pandera.DataFrameSchemaのインスタンスを作成することでスキーマが定義できます。

基本の書き方：

1. schema = pa.DataFrameSchemaとインスタンスの形で定義する。
2. columns = {...}と辞書形式で引数を持たせる。
3. 辞書の中に"列名": pa.Column(列の詳細)という形で各列の情報を与える。

python

schema = pa.DataFrameSchema(
    {
        "名前": pa.Column(
            str,
            pa.Check.str_length(
                3, 10, error="名前の長さは3文字以上10文字以下である必要があります。"
            ),
            description="顧客の名前（3～10文字）。",
        ),
        "年齢": pa.Column(
            int,
            pa.Check.ge(0, error="年齢は0以上の整数である必要があります。"),
            description="顧客の年齢（0歳以上）。",
        ),
        "メールアドレス": pa.Column(
            str,
            pa.Check.str_matches(
                r".+@.+\..+", error="メールアドレスの形式が正しくありません。"
            ),
            description="有効なメールアドレス形式。",
        ),
        "購入金額": pa.Column(
            float,
            pa.Check.gt(0, error="購入金額は正の数である必要があります。"),
            nullable=True,
            description="購入金額（円）。正の数で、欠損値も許容。",
        ),
    }
)

詳しくは後述しますが、スキーマを定義する方法にはpa.DataFrameSchemaを使用する方法とpa.SchemaModelを使用する方法があります。ここでは、pa.DataFrameSchemaを使用して説明します。

3. validateメソッドでバリデーションを行う。

以下のコードでバリデーションチェックを行います。

schema.validate(data)    

この場合は、購入金額に-500.0と正しくない値が入っているため、以下のようなエラー（pandera.error.SchemaError）が起こります。

SchemaError: Column '年齢' failed element-wise validator number 0: 年齢は0以上の整数である必要があります。 failure cases: -5

各列のスキーマ定義

dtype（第一引数）

列の要素の型を指定します。
型が異なる場合、pandera.errors.SchemaErrorを起こします。

python

schema = pa.DataFrameSchema(
    {
        "名前": pa.Column(dtype=str, description="顧客の名前（文字列）。"),
        "年齢": pa.Column(dtype=int, description="顧客の年齢（整数）。"),
        "購入金額": pa.Column(
            dtype=float,
            nullable=True,
            description="購入金額（浮動小数点数、欠損値許容）。",
        ),
    }
)

checks（第二引数）とpa.Check

引数checksにはpa.Checkインスタンスを入れ、列により詳細な条件を課します。
条件を満たさない場合、pandera.errors.SchemaErrorを起こします。

pa.Checkのクラスメソッド

よく用いられる条件は予めpa.Checkクラスメソッドとして保持されています。
同じ行に書いてあるメソッドはどちらを使用しても問題ありません。

メソッド	詳細
equal_to(n)・eq(n)	値がnに等しいか
not_equal_to(n) ・ne(n)	値がnと等しくないか
greater_than(n) ・gt(n)	値がnより大きいか
greater_than_or_equal_to(n) ・ge(n)	値がn以上か
less_than(n) ・lt(n)	値がnより小さいか
less_than_or_equal_to(n)・le(n)	値がn以下か
in_range(n, m)	値がn以上m以下か
isin(iterable)	値がiterableに含まれるか
notin(iterable)	値がiterableに入っていないか
str_matches(r"^n$")	文字列が指定した正規表現nに一致しているか
str_contains("aaa")	文字列の中にaaaが含まれているか
str_startswith("aaa")	文字列がaaaで始まるか
str_endswith("aaa")	文字列がaaaで終わるか
str_length(n)	文字列の文字数がnか
unique_values_eq(iterable)	値がiterableに含まれており、Iterable内の全ての値が列内の要素に存在するか

python

schema = pa.DataFrameSchema(
    {
        "年齢": pa.Column(
            int,
            checks=pa.Check.in_range(0, 120),
            description="年齢が0～120歳の範囲内であること。",
        )
    }
)

lambdaによる条件

上記リストにないチェックを行いたい場合は、lambdaを利用して条件を定義することも可能です。

lambdaの引数はpa.Check内の引数element_wiseをTrueにすると列の各要素（strやfloatなど）になり、Falseにすると列全体（pd.Series）になります。

python

schema = pa.DataFrameSchema(
    {
        "購入金額": pa.Column(
            float,
            pa.Check(lambda x: x % 100 == 0, element_wise=True),
            description="購入金額が100円単位であること。",
        )
    }
)

複数条件の組み合わせ

複数条件を論理積（かつ）の形で課したい場合は、引数checksをpa.Checkのlistの形で与えることで可能です。

import math


# 素数判定関数
def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True


# スキーマ定義
schema = pa.DataFrameSchema(
    {
        "値": pa.Column(
            int,
            checks=[
                pa.Check.in_range(0, 10000, error="値が0以上10000以下ではありません"),
                pa.Check(
                    lambda x: is_prime(x),
                    element_wise=True,
                    error="値は素数である必要があります",
                ),
            ],
            description="0以上10000以下の素数。",
        )
    }
)

groupbyとgroups

「年齢が18歳以下だったら購入金額は1,000円以下」のように、別の列の値によって条件が異なる場合がある状況も存在するでしょう。そのような場合に便利なのが、pa.Checkの引数groupbyです。groupbyに列名を入力すると、その列名でグルーピングを行います。
groupbyを指定した場合、pa.Check内のlambdaの引数の型は{groupbyの列の値: 該当する列の値のSeries}という辞書の形になります。デフォルトでは辞書には全てのgroupbyの列の値に入りますが、groupsで特定の値を指定するとその値をkeyに持つ辞書だけ残ります。

# スキーマ定義
schema = pa.DataFrameSchema(
    {
        "購入金額": pa.Column(
            int,
            [
                # groupbyを使用し、18歳以下の条件を検証
                pa.Check(
                    lambda g: all(g[True] <= 1000),
                    groupby=lambda df: df.assign(
                        under_18=lambda d: d["年齢"] <= 18
                    ).groupby("under_18"),
                    error="18歳以下の購入金額は1,000円以下である必要があります。",
                )
            ],
            description="購入金額（円）",
        ),
        "年齢": pa.Column(int, pa.Check.gt(0), description="年齢（正の整数）"),
    }
)

# サンプルデータ
data = pd.DataFrame({"年齢": [17, 18, 19, 20], "購入金額": [800, 1100, 1200, 1500]})

# バリデーション実行
schema.validate(data)

SchemaError: Column '購入金額' failed series or dataframe validator 0: <Check <lambda>: 18歳以下の購入金額は1,000円以下である必要があります。>

error

checksでは条件を自由に設定できるため、チームメンバーがpandera.error.SchemaErrorに遭遇した際、なぜエラーが発生するのかがわからなくなってしまう可能性があります。そこで、pa.Checkの引数errorに、表示されるエラー文を入れておくことで、何が原因なのかをログに記載させるようにしましょう。

# スキーマ定義
schema = pa.DataFrameSchema(
    {
        "年齢": pa.Column(
            int,
            pa.Check.in_range(0, 120, error="年齢は0～120の範囲でなければなりません"),
        ),
        "メールアドレス": pa.Column(
            str,
            pa.Check.str_matches(
                r".+@.+\..+", error="メールアドレス形式が正しくありません"
            ),
        ),
    }
)

# サンプルデータ
data = pd.DataFrame(
    {"年齢": [-1, 130], "メールアドレス": ["invalid_email", "[e-mail]"]}
)

# バリデーション実行
schema.validate(data)

SchemaError: Column '年齢' failed element-wise validator number 0: 年齢は0～120の範囲でなければなりません failure cases: -1, 130

descriptionで列の説明を入れておこう

pa.Columnの引数にはdescriptionという引数が存在します。これはスキーマの定義をJSON形式で保管しておく際に用いられるもので、データフレームの評価には関係ありませんが、どのような列なのかチームメンバーが分かるように、列の説明を入れておきましょう。

indexの定義

indexも列と同様スキーマを定義することができます。pa.DataFrameSchemaの引数indexにクラスpa.Indexを入れ、列と同様にスキーマ定義ができます。
また、pa.MultiIndexを入れれば、マルチインデックスにも対応可能です。

# サンプルデータ
data = pd.DataFrame({
    "顧客ID": ["1", "1", "2", "3", "3", "3"],
    "購入金額": [100, 200, 300, 400, 500, 600]
})

# groupby 後のデータ
grouped_data = data.groupby("顧客ID").agg(購入回数=("購入金額", "size"))

# スキーマ定義
schema = pa.DataFrameSchema(
    columns={
        "購入回数": pa.Column(
            int,
            pa.Check.gt(0, error="購入回数は正の整数である必要があります。"),
            description="顧客の購入回数。"
        )
    },
    index=pa.Index(
        str,
        pa.Check.str_matches(
            r"^[0-9]+$", error="顧客IDは数値のみで構成されている必要があります。"
        ),
        name="顧客ID",
        description="顧客を一意に識別するID。"
    )
)

# バリデーション実行
schema.validate(grouped_data)

その他検証に用いられる、bool値の引数

その他にも使用機会の多い、pa.Column内のチェックに関する引数を紹介します。

引数	デフォルト	説明
nullable	False	Trueにすると列に空欄（np.nan）があっても、型関係なくErrorを起こさなくなります。 Falseでは空欄にも検証が入り、型などが異なればErrorを起こします。
unique	False	Trueにすると列に重複する値があると、Errorを起こします。
required	True	Trueにすると、その列がデータにないとErrorを起こします。 FalseにするとErrorを起こさず、列が存在している場合だけ検証されます。

列の追加

プロセスの中で、列が追加されることは多々あります。その時、pa.DataFrameSchemaのメソッド.add_columnsを使えば、いちいち全て再定義せずに、列を追加することができます。

# 初期スキーマ定義
schema = pa.DataFrameSchema(
    {
        "顧客ID": pa.Column(
            str
            pa.Check.str_matches(
                r"^[0-9]+$", error="顧客IDは数値のみで構成されている必要があります。"
            ),
        ),
        "購入金額": pa.Column(
            float, pa.Check.gt(0), description="購入金額（正の数値）"
        ),
    }
)

# 後から新たな列「商品数」を追加
schema = schema.add_columns(
    {"商品数": pa.Column(int, pa.Check.ge(1), description="商品数（1以上）")}
)

# サンプルデータ
data = pd.DataFrame({"顧客ID": ["1", "2"], "購入金額": [100.0, 200.0], "商品数": [2, 3]})

# バリデーション実行
schema.validate(data)

DataFrameの検証

先述のとおり、スキーマインスタンス.validate(データフレーム)とすれば検証が行えます。

python

schema.validate(data)

このメソッドの返り値はデータフレームそのものなので、__init__の中に入れて、インスタンス作成時にバリデーションを行うことも可能です。

python

class SampleDatas:
    def __init__(self, input_data):
        self.data = schema.validate(input_data)

lazy

pa.DataFrameSchema.validationの引数lazyをTrueにすると、データフレームの全ての列に検証を行い、pandera.errors.SchemaErrorをまとめて出力させることができます。この時、validateだけではログには最初のErrorだけしか返ってこないため、try-exceptブロック等を使い、エラー文をlogger.errorなどで出力させる必要があります。

python

import pandas as pd
import pandera as pa
import logging

# ログの設定
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

# スキーマ定義
schema = pa.DataFrameSchema(
    {
        "顧客ID": pa.Column(
            str,
            pa.Check.str_matches(
            r"^[0-9]+$", error="顧客IDは数値のみで構成されている必要があります。"
            )
        ),
        "購入金額": pa.Column(
            float, pa.Check.gt(0, error="購入金額は正の数値でなければなりません")
        ),
        "商品数": pa.Column(
            int, pa.Check.ge(1, error="商品数は1以上でなければなりません")
        ),
    }
)

# サンプルデータ（複数のエラーを含む）
data = pd.DataFrame(
    {
        "顧客ID": ["1", "2", "3"],
        "購入金額": [100.0, -200.0, 0],  # -200.0 と 0 がエラー
        "商品数": [0, 2, -1],  # 0 と -1 がエラー
    }
)

# 検証
try:
    if __name__ == "__main__":
        schema.validate(data, lazy=True)  # lazy=True ですべてのエラーを収集
except pa.errors.SchemaErrors as e:
    logger.error("バリデーションエラーが発生しました:")
    for _, failure in e.failure_cases.iterrows():
        logger.error(
            f"行: {int(failure['index'])}, "
            f"列: {failure['column']}, "
            f"条件: {failure['check']}, "
            f"値: {failure['failure_case']}"
        )

ERROR:__main__:バリデーションエラーが発生しました:
ERROR:__main__:行: 1, 列: 購入金額, 条件: 購入金額は正の数値でなければなりません, 値: -200.0
ERROR:__main__:行: 2, 列: 購入金額, 条件: 購入金額は正の数値でなければなりません, 値: 0.0
ERROR:__main__:行: 0, 列: 商品数, 条件: 商品数は1以上でなければなりません, 値: 0.0
ERROR:__main__:行: 2, 列: 商品数, 条件: 商品数は1以上でなければなりません, 値: -1.0

上記の例では使用しておりませんが、エラー詳細はe.failure_casesというDataFrameで取得できます。そのカラム例は以下の通りです。

列名	説明
schema_context	エラーが発生したコンテキスト（Columnなど）
column	エラーが発生した列名
check	失敗したバリデーション条件
failure_case	バリデーションに失敗した値
index	元のデータフレームでの行インデックス

これにより、どの行・列・値が問題だったかを詳細に把握できます。

デコレータによる検証

関数やメソッドの中でバリデーションを行う場合、場合によってはデコレータで自動的に検証できます。

デコレータ	詳細
@pa.check_input(schema)	関数の引数（インプット）のdfにバリデーションを行います。
@pa.check_output(schema)	関数の返り値（アウトプット）のdfにバリデーションを行います。
@pa.check_io(inputs=schema1, out=schema2)	関数の引数（インプット）のdfと返り値（アウトプット）のdfにバリデーションを行います。

python

# 入力スキーマ定義
input_schema = pa.DataFrameSchema(
    {
        "顧客ID": pa.Column(
            str,
            pa.Check.str_matches(
                r"^[0-9]+$", error="顧客IDは数値のみで構成されている必要があります。"
            ),
            description="顧客を識別する正の整数ID",
        ),
        "購入金額": pa.Column(
            float,
            pa.Check.ge(0, error="購入金額は0以上の整数である必要があります"),
            description="購入金額（0以上の数値）",
        ),
    }
)

# 出力スキーマ定義
output_schema = pa.DataFrameSchema(
    {
        "顧客ID": pa.Column(
            str,
            pa.Check.str_matches(
                r"^[0-9]+$", error="顧客IDは数値のみで構成されている必要があります。"
            ),
        ),
        "総購入金額": pa.Column(
            float,
            pa.Check.gt(0, error="総購入金額は0以上の整数である必要があります"),
            description="顧客の総購入金額（正の数値）",
        ),
    }
)


@pa.check_io(data=input_schema, out=output_schema)
def calculate_total_purchase(data: pd.DataFrame):
    return data.groupby("顧客ID").agg(総購入金額=("購入金額", "sum")).reset_index()


# サンプルデータ
data = pd.DataFrame(
    {
        "顧客ID": ["1", "2", "2", "3"],
        "購入金額": [100.0, 200.0, 300.0, -50.0],
    }
)

try:
    result = calculate_total_purchase(data)
    print("処理結果:")
    print(result)
except pa.errors.SchemaErrors as e:
    print("バリデーションエラーが発生しました。")
    print(e.failure_cases)

DataFrameの自動整形

panderaはデータフレームの検証を行い、合致していなかったらエラーを起こすだけでなく、ガッチしていない部分を自動修正させることもできます。

自動整形を利用すると、データフレームがpanderaのスキーマに基づいて自動的に修正されます。そのため、無理やりな修正や意図しない修正により、後々のプロセスで重大なエラーを起こす可能性があります。自動整形を利用する際は以下の点に注意してください。

• 元のデータフレームをバックアップする
  自動整形を適用すると、元のデータの情報が失われます。何かあった時に整形前のデータを確認できるようにしておきましょう。
• スキーマの条件を明確に定義する
  自動修正が必要な列や型、許容値を厳密に指定し、意図しない変更が起きないようにしましょう。
• データフレームとスキーマが一致しない原因をする
  データフレームに重大な誤りがあっても、自動整形を行う前に検証だけを行い、自動整形を行う妥当性を確認しましょう。
• 結果を必ず確認する
  本当に予期せぬ修正が起こっていないか、整形後のデータが期待通りになっているか確認しましょう。

coerce

pa.Columnの引数coerceをTrueにすると、スキーマの型になるように値を自動修正します。修正不可能な場合はpandera.errors.SchemaErrorを起こします。

python

schema = pa.DataFrameSchema(
    {"年齢": pa.Column(int, coerce=True), "購入金額": pa.Column(float, coerce=True)}
)

data = pd.DataFrame({"年齢": ["25", "30"], "購入金額": ["1000.0", "1500.5"]})

validated_data = schema.validate(data)
print(validated_data.dtypes)

年齢        int64
購入金額    float64
dtype: object

parsers

pa.Columnの引数parserにpa.Parserオブジェクトを入れ、その中にpd.Seriesを引数及び返り値とする関数を入れると、列を関数に基づいて変化させます。欠損値埋めや文字列のトリミングなどに便利です。

python

# スキーマ定義
schema = pa.DataFrameSchema(
    {
        "名前": pa.Column(
            str,
            parsers=pa.Parser(lambda s: s.str.strip().str.title()),
            description="名前（余分なスペースを除去し、タイトルケースに変換）",
        )
    }
)

# サンプルデータ
data = pd.DataFrame({"名前": ["  yamada taro  ", "suzuki HANako "]})

# バリデーション実行
validated_data = schema.validate(data)
print(validated_data)

              名前
0    Yamada Taro
1  Suzuki Hanako

add_missing_columnsとdefault

pa.Columnの引数add_missing_columnsをTrueとすると、該当する列がデータフレームに存在しない場合、列を自動で追加します。追加される際の列の値はpa.Columnの引数defaultに設定することができます。

python

schema = pa.DataFrameSchema(
    {
        "名前": pa.Column(str, description="顧客の名前"),
        "年齢": pa.Column(int, default=0, description="年齢（デフォルト値は0）"),
    },
    add_missing_columns=True,
)

data = pd.DataFrame({"名前": ["山田太郎"]})

validated_data = schema.validate(data)
print(validated_data)

     名前  年齢
0  山田太郎   0

追記1 DataFrameModel　v.s.　DataFrameSchema

panderaにてバリデーションを行うにあたって、まずはスキーマ定義を行うわけですが、DataFrameの定義はDataFrameModelを継承したclassとして定義する方法とDataFrameSchemaクラスのインスタンスとして定義する方法があります。
それぞれの特徴は後述しますが、共同開発を行うならば予め宣言された変数で列名を定義できるDataFrameSchemaによる定義の方が良いのではないかと考えております。

DataFrameSchema

これまでのように、DataFrameSchemaのインスタンスとしてスキーマを定義します。各列の情報はcolumns引数の中で{列名: 列の情報}と辞書の形で定義します。

• メリット
  • 変数や定数で列名を定義しても、その変数や定数を使ってスキーマ定義が行える。
• デメリット
  • クラスの形で定義しているわけではなく、見た目の統一感がある保証がないため、見辛い場合がある。

python

# 列名を変数で定義
CUSTOMER_ID = "顧客ID"
PURCHASE_AMOUNT = "購入金額"
ITEM_COUNT = "商品数"

# スキーマ定義
schema = pa.DataFrameSchema(
    {
        CUSTOMER_ID: pa.Column(
            str,
            pa.Check.str_matches(r"^[0-9]+$"),
            description="顧客を識別する正の整数ID"
        ),
        PURCHASE_AMOUNT: pa.Column(
            float, pa.Check.gt(0), description="購入金額（正の数）"
        ),
        ITEM_COUNT: pa.Column(int, pa.Check.ge(1), description="商品数（1以上）"),
    }
)

# サンプルデータ
data = pd.DataFrame(
    {
        CUSTOMER_ID: ["1", "2", "3"],
        PURCHASE_AMOUNT: [100.0, 200.0, 300.0],
        ITEM_COUNT: [1, 2, 3],
    }
)

# バリデーション実行
schema.validate(data)

DataFrameModel

DataFrameModelの子クラスとしてスキーマを定義します。各列の情報は列名: 列の情報とクラス変数で定義します。

• メリット
  • クラス変数として列名が置かれているため、読みやすい。
  • デコレータ@pa.check_typesと関数の引数の型をpa.typing.DataFrame[Schema]とすることで直感的な検証ができる。
  • スキーマのクラスをさらに継承したり、クラス変数を追加したりすれば、列を追加してスキーマ定義できるため、直感的にカスタマイズがしやすい。
  • チェックや修正のための関数をクラスメソッドとして定義できる。
• デメリット
  • 列名を変数や定数で定義しても、それをそのまま使うことができない。

python

from pandera.typing import DataFrame


class PurchaseDataModel(pa.DataFrameModel):
    顧客ID: str = pa.Field(
        str_matches=r"^[0-9]+$", description="顧客を識別する正の整数ID"
    )
    購入金額: float = pa.Field(gt=0, description="購入金額（正の数）")
    商品数: int = pa.Field(ge=1, description="商品数（1以上）")


# サンプルデータ
data = pd.DataFrame(
    {"顧客ID": ["1", "2", "3"], "購入金額": [100.0, 200.0, 300.0], "商品数": [1, 2, 3]}
)


# 関数でのスキーマ検証
@pa.check_types
def process_purchase_data(
    data: DataFrame[PurchaseDataModel],
) -> DataFrame[PurchaseDataModel]:
    # データをそのまま返す処理（例: データ処理の前後でスキーマをチェック）
    return data


# 関数呼び出しでバリデーション
data = process_purchase_data(data)

pa.Column内の引数nameを使うことで、pa.DataFrameModelでも変数で列名を使用することが可能ではあります。
しかし、かえって列名がぱっと見でわかりにくくなってしまうため、やはりpa.DataFrameSchemaを使った方が良いでしょう。

追記2 その他オプション

strict

基本的にvalidationのときは記載されている列に関してのスキーマだけ行い、他に列が存在しても特に影響はありません。しかし、異なるデータフレームをmergeするときなど、列の存在が悪影響を及ぼすことがあります。スキーマにて定義された列以外の列の存在の有無に関する操作が、pa.DataFrameSchema内の引数strictです。

• strict=Trueとすると、スキーマに定義されていない列が存在してている場合、pandera.error.SchemaErrorが発生します。
• strict="filter"とすると、スキーマに定義されていない列が存在してている場合、その列は自動的に削除されます。

python

# スキーマ定義　strict="filter" を使用
schema = pa.DataFrameSchema(
    {
        "顧客ID": pa.Column(
            str,
            pa.Check.str_matches(r"^[0-9]+$"),
            description="顧客を識別する正の整数ID",
        ),
        "購入金額": pa.Column(
            float, pa.Check.gt(0), description="購入金額（正の数値）"
        ),
        "商品数": pa.Column(int, pa.Check.ge(1), description="商品数（1以上）"),
    },
    strict="filter",  # 未定義の列を自動的に削除
)

# サンプルデータ
data = pd.DataFrame(
    {
        "顧客ID": [1, 2],
        "購入金額": [100.0, 200.0],
        "商品数": [1, 2],
        "EXTRA_COLUMN": ["extra1", "extra2"],  # 定義されていない列
    }
)

# バリデーション実行
print(schema.validate(data).columns)

Index(['顧客ID', '購入金額', '商品数'], dtype='object')

ordered

基本的にvalidationのときはデータフレームの列の順序は関係なく検証が行われます。機械学習など、データの列の順序が重要になる場合、pa.DataFrameSchema内の引数orderedをTrueにすると、データフレームの列の順序がスキーマ通りになっていない場合、pandera.error.SchemaErrorが発生します。

python

# スキーマ定義
schema = pa.DataFrameSchema(
    {
        "顧客ID": pa.Column(str, pa.Check.str_matches(r"^[0-9]+$"),
        "購入金額": pa.Column(float, pa.Check.gt(0)),
        "商品数": pa.Column(int, pa.Check.ge(1)),
    },
    ordered=True,  # 列の順序がスキーマ通りでないとエラー
)

# サンプルデータ
data = pd.DataFrame({"購入金額": [100.0, 200.0], "顧客ID": ["1", "2"], "商品数": [1, 2]})

# バリデーション実行
schema.validate(data)

SchemaError: column '購入金額' out-of-order

おわりに

panderaの一番のメリットとして、列名をバリデーションのついでに記載できること、バリデーションが直感的なことだと考えております。作業の際も記載されたスキーマをもとに、どのような列があったか自分で再確認する、まるでメモ代わりのような扱いをすることが多く、便利でした。そのため、regrexという引数で複数の列に同時にスキーマを定義したり、データフレーム全体に型指定などを行う機能もあったのですが、ここではあえて紹介しませんでした。
この記事を執筆するにあたって、panderaの公式ページを隅々まで読んで理解しました。groupbyなどは記載している記事が少なく、使い方をChatGPTも間違えるほどだったのですが、この記事が参考例になればと思います。
また、panderaにはまだ修正すべき点があるように感じております。pa.Check.unique_values_eqの引数の型指定がなぜかstrだったり、pa.Columnの引数drop_invalid_rowsが何故かなんの意味も為していなかったり…。これらのバグに対して、余裕があれば修正してプルリクが送れるほどスキルを磨けたら、と目標の一つに掲げることもできました。

まとめ

この記事のポイント

• panderaを使えば、DataFrameの列定義とバリデーションを同時に行える。
• pa.DataFrameSchemaやpa.DataFrameModelを用いて、データ構造と品質を明示的に定義可能。
• checksやgroupbyを組み合わせて、複雑な条件付バリデーションも行える。
• lazy=Trueやstrict、orderedなどのオプションで、エラー収集や列順序制約、未定義列処理を柔軟に制御。
• 自動整形(coerce, parserなど)機能により、データを期待形へ近づけることも可能。

導入の流れ(例):

1. スキーマ定義する。
2. プロセスの最初と最後にvalidateでデータフレームの検証を行う。
3. エラー発生時はloggerで詳細な情報を記録し、修正やフィードバックサイクルを回す。
4. 必要に応じて、問題なさそうであればcoerceなどで修正し、再度バリデーション。
5. 成功すれば次のプロセスへ渡す。