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

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

データフレームはどのような列・値が存在していても一律で`pd.DataFrame`として扱うことが多く、その中身はコードを隅々まで読まないとわからないブラックボックス状態になってしまいます。
- **チームメンバーに処理を引き継ぐ場合、伝えることが難しい。**
    dfをインプットとして用いるとき、大抵列名を指定してそのスライスを見ることが多いです。インプットにするならばどのような列があるのか、その値は相応しいものなのかがわからなくなります。チームメンバーから「このdfにはどのような列がありますか？」「必要な値はなんていう列の中に入っていますか？」とチームメンバーから質問されることも多くありました。
    <br>

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

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

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


https://pandera.readthedocs.io/en/stable/index.html

# panderaの基本

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

### panderaとは

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

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

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

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

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


### panderaのインストール

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

### 使い方要約

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

```python:python
import pandas as pd

data = pd.DataFrame(
    {
        "名前": ["山田太郎", "鈴木花子", "佐藤次郎"],
        "年齢": [25, 30, -5],
        "メールアドレス": ["yamada@example.com", "suzuki@example", "sato@example.com"],
        "購入金額": [1200.5, None, -500.0],
    }
)
```
以降、`import pandas as pd`を前提として紹介していきます。

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

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

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

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

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

```python: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="購入金額（円）。正の数で、欠損値も許容。",
        ),
    }
)
```

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

#### 3. **validateメソッドでバリデーションを行う。**
以下のコードでバリデーションチェックを行います。

```python
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: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: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:python
schema = pa.DataFrameSchema(
    {
        "購入金額": pa.Column(
            float,
            pa.Check(lambda x: x % 100 == 0, element_wise=True),
            description="購入金額が100円単位であること。",
        )
    }
)
```

#### 複数条件の組み合わせ

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

```python
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に持つ辞書だけ残ります。

```python
# スキーマ定義
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`に、表示されるエラー文を入れておくことで、何が原因なのかをログに記載させるようにしましょう。

```python
# スキーマ定義
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", "valid@example.com"]}
)

# バリデーション実行
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`を入れれば、マルチインデックスにも対応可能です。


```python
# サンプルデータ
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を起こさなくなります。 <br> `False`では空欄にも検証が入り、型などが異なればErrorを起こします。|
| `unique` | `False` | `True`にすると列に重複する値があると、Errorを起こします。|
| `required` | `True` | `True`にすると、その列がデータにないとErrorを起こします。<br>`False`にするとErrorを起こさず、列が存在している場合だけ検証されます。 |

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

```python
# 初期スキーマ定義
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:python
schema.validate(data)
```
このメソッドの返り値はデータフレームそのものなので、__init__の中に入れて、インスタンス作成時にバリデーションを行うことも可能です。

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

:::note warn
自動整形を利用すると、データフレームがpanderaのスキーマに基づいて自動的に修正されます。そのため、無理やりな修正や意図しない修正により、後々のプロセスで重大なエラーを起こす可能性があります。自動整形を利用する際は以下の点に注意してください。
- **元のデータフレームをバックアップする**
    自動整形を適用すると、元のデータの情報が失われます。何かあった時に整形前のデータを確認できるようにしておきましょう。
- **スキーマの条件を明確に定義する**
    自動修正が必要な列や型、許容値を厳密に指定し、意図しない変更が起きないようにしましょう。
- **データフレームとスキーマが一致しない原因をする**
    データフレームに重大な誤りがあっても、自動整形を行う前に検証だけを行い、自動整形を行う妥当性を確認しましょう。
- **結果を必ず確認する**
    本当に予期せぬ修正が起こっていないか、整形後のデータが期待通りになっているか確認しましょう。
:::

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

```python: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: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: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: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: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)
```
    


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


# 追記2 その他オプション

### `strict`
基本的にvalidationのときは記載されている列に関してのスキーマだけ行い、他に列が存在しても特に影響はありません。しかし、異なるデータフレームをmergeするときなど、列の存在が悪影響を及ぼすことがあります。スキーマにて定義された列以外の列の存在の有無に関する操作が、`pa.DataFrameSchema`内の引数`strict`です。
- `strict=True`とすると、スキーマに定義されていない列が存在してている場合、`pandera.error.SchemaError`が発生します。
- `strict="filter"`とすると、スキーマに定義されていない列が存在してている場合、その列は自動的に削除されます。

```python: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: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. 成功すれば次のプロセスへ渡す。