Pandera 로 Schema 검증

제 몫 하는 가장 가벼운 검증

Pandera 는 Pandas (그리고 Polars, Modin, Dask) 를 위한 Python 스타일 schema 검증 라이브러리야. 쓰는 법의 뼈대는 이래: 기대하는 schema 를 선언하고, schema.validate(df) 를 부르면, 검증된 DataFrame 을 받거나 모든 위반을 조목조목 나열한 예외를 받아. Silent corruption 을 시끄럽고 디버깅 가능한 실패로 바꾸는 방어 수단 중 제일 싼 거야.

받는 것

Type 체크 — column dtype 이 선언한 type 과 맞는지 확인하고, coerce=True 면 자동 cast 까지.
제약 체크 — unique, non-null, regex, range, custom predicate.
Lazy 수집 — schema.validate(df, lazy=True) 가 첫 위반에서 멈추지 않고 DataFrame 전체의 위반을 다 모아 줘.
DataFrameModel 클래스 — Pydantic 모델처럼 선언형으로 schema 정의.
Polars 지원 — Pandera 0.20+ 부터는 같은 schema 를 Polars DataFrame 에도 그대로 적용.

Code

가장 유용한 4가지 제약 type 의 Pandera schema·python

import pandas as pd
import pandera.pandas as pa
from pandera.errors import SchemaErrors

schema = pa.DataFrameSchema({
    'order_id':    pa.Column(str, unique=True,
                              checks=pa.Check.str_matches(r'^O\d{6}$')),
    'customer_id': pa.Column(str, checks=pa.Check.str_matches(r'^C\d{4,8}$')),
    'order_date':  pa.Column(pa.DateTime, coerce=True,
                              checks=pa.Check.greater_than('2020-01-01')),
    'amount_usd':  pa.Column(float, coerce=True,
                              checks=[pa.Check.ge(0), pa.Check.le(1_000_000)]),
    'status':      pa.Column(str, checks=pa.Check.isin(['pending', 'completed', 'cancelled'])),
    'country':     pa.Column(str, nullable=True,
                              checks=pa.Check.str_matches(r'^[A-Z]{2}$')),
})

try:
    clean = schema.validate(raw_df, lazy=True)        # 모든 에러 수집
except SchemaErrors as e:
    # e.failure_cases 가 모든 fail 한 row × check 의 DataFrame
    print(e.failure_cases.head(20))
    raise

DataFrameModel — 선언적, IDE 친화 schema·python

import pandera.pandas as pa
import pandas as pd

class OrderSchema(pa.DataFrameModel):
    order_id:    pa.typing.Series[str]   = pa.Field(unique=True, str_matches=r'^O\d{6}$')
    customer_id: pa.typing.Series[str]   = pa.Field(str_matches=r'^C\d{4,8}$')
    order_date:  pa.typing.Series[pd.Timestamp] = pa.Field(coerce=True)
    amount_usd:  pa.typing.Series[float] = pa.Field(coerce=True, ge=0, le=1_000_000)
    status:      pa.typing.Series[str]   = pa.Field(isin=['pending', 'completed', 'cancelled'])

    class Config:
        strict = True       # 추가 column raise
        coerce = True       # 가능한 데서 자동 cast

clean = OrderSchema.validate(raw_df, lazy=True)

Exercise

본인이 작업한 DataFrame 에 대해 Pandera DataFrameModel 을 정의해 봐. Regex 체크, range 체크, isin 체크를 최소 하나씩 넣어. 실제 데이터에 lazy=True 로 검증을 돌리고 e.failure_cases 에서 뭐가 걸렸는지 봐 — 깨끗하다고 믿었던 데이터에서도 보통 한 가지 이상은 튀어나와.

제 몫 하는 가장 가벼운 검증

받는 것

Code

External links

Exercise

Progress

댓글 0