Pandera: Validasi Data Statistik untuk DataFrame pandas dan Polars

Pipeline data sering gagal tanpa suara. Sebuah kolom yang seharusnya tidak pernah negatif lolos begitu saja, salah ketik pada kategori merusak join di tahap berikutnya, dan Anda baru menyadarinya ketika dashboard terlihat keliru tiga hari kemudian. Pandera memungkinkan Anda mendeklarasikan bentuk DataFrame yang seharusnya langsung di dalam Python dan menegakkan aturan itu tepat di tempat data mengalir. Tutorial ini membahas Pandera dari instalasi pertama hingga integrasi pipeline produksi, dengan dataset transaksi penjualan sebagai contoh yang dipakai sepanjang artikel.

Apa Itu Pandera

Pandera adalah pustaka validasi data untuk data tabular. Anda mendefinisikan sebuah skema, yaitu deskripsi kolom, tipe datanya, dan batasan yang harus dipenuhi, lalu Anda memvalidasi sebuah DataFrame terhadap skema tersebut. Jika data sesuai, DataFrame dikembalikan tanpa perubahan. Jika tidak, Pandera memunculkan error yang memberi tahu Anda persis baris dan kolom mana yang gagal beserta alasannya.

Ide utamanya adalah skema merupakan objek Python biasa. Skema berada di dalam basis kode Anda, berdampingan dengan fungsi yang menghasilkan dan mengonsumsi data. Pendekatan ini kadang disebut "schema as code": tidak ada file konfigurasi terpisah, tidak ada layanan validasi eksternal, dan tidak ada YAML yang harus dijaga agar tetap sinkron. Anda mengimpor skema sama seperti mengimpor modul lain.

Pandera awalnya berfokus pada pandas dan kini mendukung beberapa backend, termasuk Polars dan PySpark, melalui model skema bersama. Logika validasi yang Anda tulis sebagian besar sama, terlepas dari mesin DataFrame yang ada di baliknya.

Kapan Memakai Pandera dibanding Great Expectations

Kedua pustaka memvalidasi data tabular, tetapi keduanya menyasar alur kerja yang berbeda.

Great Expectations adalah kerangka kerja yang lebih berat. Ia memelihara sebuah data context, menghasilkan dokumentasi data berbentuk HTML, menyimpan hasil validasi, dan dirancang dengan gagasan platform kualitas data terpusat yang dipakai bersama oleh analis dan engineer. Ini cocok ketika Anda membutuhkan laporan validasi yang dapat diaudit, katalog ekspektasi, dan perkakas yang juga digunakan oleh non-pengembang.

Pandera ringan dan mengutamakan kode. Ia menambahkan validasi secara inline di proses Python yang sama dengan proses yang mentransformasi data. Tidak ada context yang perlu dikonfigurasi dan tidak ada penyimpanan artefak. Anda memilih Pandera ketika menginginkan asersi yang berada di dalam fungsi ETL, batas fungsi yang diperiksa tipenya, dan validasi yang berjalan sebagai bagian normal dari pipeline tanpa infrastruktur tambahan.

Aturan praktisnya: jika validasi adalah urusan pengembang yang menyatu dengan kode, pilih Pandera. Jika validasi adalah urusan organisasi dengan dokumentasi dan pelaporan bersama, Great Expectations sepadan dengan bobotnya. Keduanya tidak saling meniadakan; sebagian tim memakai Pandera untuk pemeriksaan inline yang cepat dan Great Expectations untuk kontrak yang terdokumentasi.

Instalasi

Pasang paket inti dengan pip.

pip install pandera

Dukungan backend dan fitur opsional didistribusikan sebagai extras. Pasang hanya yang Anda butuhkan.

# Dukungan Polars
pip install 'pandera[polars]'

Dukungan PySpark
pip install 'pandera[pyspark]'

Pemeriksaan statistik berbasis hypothesis
pip install 'pandera[hypotheses]'

Semuanya
pip install 'pandera[all]'

Verifikasi instalasi dan periksa versinya.

import pandera as pa

print(pa.version)

Skema Pertama dengan DataFrameSchema

API berbasis objek berpusat pada DataFrameSchema, yang menyimpan pemetaan nama kolom ke definisi Column. Berikut sebuah skema untuk tabel transaksi penjualan.

import pandas as pd
import pandera as pa
from pandera import Column, Check, DataFrameSchema

schema = DataFrameSchema(
    {
        "transactionid": Column(int, unique=True),

        "product": Column(str),
        "category": Column(str),
        "quantity": Column(int, Check.greaterthan(0)),
        "unitprice": Column(float, Check.greaterthanorequalto(0)),

        "region": Column(str),
        "customeremail": Column(str, nullable=True),
    },
    strict=True,
    coerce=True,
)

Setiap Column menerima sebuah dtype dan batasan opsional. Beberapa parameter melakukan sebagian besar pekerjaan:

• nullable mengizinkan nilai kosong pada kolom. Nilai bawaannya False, jadi secara default kolom tidak boleh berisi null.
• unique mengharuskan setiap nilai pada kolom berbeda satu sama lain.
• coerce mengonversi kolom ke dtype yang dideklarasikan sebelum validasi, alih-alih gagal ketika tipe tidak cocok persis. Ini bisa diatur per kolom atau untuk seluruh skema.

Argumen strict=True pada skema menolak DataFrame mana pun yang memuat kolom yang tidak dideklarasikan di skema. Tanpa itu, kolom tambahan diabaikan.

Validasi sebuah DataFrame dengan memanggil schema.validate atau cukup memanggil skemanya.

df = pd.DataFrame(
    {
        "transactionid": [1, 2, 3],

        "product": ["Keyboard", "Mouse", "Monitor"],
        "category": ["Peripherals", "Peripherals", "Displays"],
        "quantity": [2, 1, 1],
        "unitprice": [29.99, 14.50, 199.00],
        "region": ["West", "East", "West"],
        "customeremail": ["a@example.com", None, "c@example.com"],

    }
)

validated = schema.validate(df)

schema = DataFrameSchema(
    {
        "quantity": Column(int, Check.greaterthan(0)),
        "discount": Column(float, Check.inrange(0.0, 1.0)),

        "category": Column(
            str, Check.isin(["Peripherals", "Displays", "Cables", "Audio"])
        ),
        "sku": Column(str, Check.strmatches(r"^SKU-\d{6}$")),
        "rating": Column(int, Check.lessthanorequalto(5)),
    }
)

Bawaan yang sering dipakai meliputi greaterthan, greaterthanorequalto, lessthan, lessthanorequalto, inrange, isin, notin, strmatches, strcontains, strlength, dan uniquevalueseq. Anda dapat memberikan daftar pemeriksaan ke satu kolom ketika lebih dari satu batasan berlaku.

Column(
    float,
    checks=[
        Check.greaterthan(0),

        Check.lessthan(100000),
    ],
)

Pemeriksaan Kustom dengan Lambda

Ketika bawaan tidak cocok, tulis Check kustom. Bentuk paling sederhana menerima sebuah fungsi yang menerima kolom sebagai Series pandas dan mengembalikan Series boolean.

schema = DataFrameSchema(
    {
        "unitprice": Column(

            float,
            Check(lambda s: s.round(2).eq(s).all(), error="harga maksimal 2 desimal"),
        ),
        "quantity": Column(
            int,
            Check(lambda s: s % 1 == 0, error="quantity harus bilangan bulat"),
        ),
    }
)

Check(lambda s: s.mean() < 1000, error="rata-rata unit price terlalu tinggi")

Kadang lebih jelas memikirkan satu nilai pada satu waktu. Setel elementwise=True dan fungsi akan menerima nilai skalar satu per satu.

# Vektorisasi: bekerja pada seluruh Series sekaligus (lebih disarankan untuk kecepatan)
Check(lambda s: s.str.startswith("SKU-"))

Element-wise: dipanggil sekali per nilai, menerima satu skalar
Check(lambda v: v.startswith("SKU-"), elementwise=True)

Utamakan pemeriksaan vektorisasi demi performa. Gunakan elementwise hanya ketika logika per baris sulit divektorisasi, misalnya mengurai string kompleks dengan try/except.

API Berbasis Kelas dengan DataFrameModel

Untuk proyek yang lebih besar, API berbasis kelas lebih mudah dibaca dan terintegrasi dengan type hint. Anda membuat subclass DataFrameModel dan mendeklarasikan kolom sebagai atribut bertipe menggunakan Series dan Field.

import pandera as pa
from pandera.typing import Series


class TransactionSchema(pa.DataFrameModel):
    transactionid: Series[int] = pa.Field(unique=True)

    product: Series[str]
    category: Series[str] = pa.Field(isin=["Peripherals", "Displays", "Cables", "Audio"])
    quantity: Series[int] = pa.Field(gt=0)
    unitprice: Series[float] = pa.Field(ge=0)
    region: Series[str] = pa.Field(isin=["West", "East", "North", "South"])
    customeremail: Series[str] = pa.Field(nullable=True)


    class Config:
        strict = True
        coerce = True

validated = TransactionSchema.validate(df)

Anda juga dapat melampirkan pemeriksaan kustom sebagai metode menggunakan dekorator @pa.check dan @pa.dataframecheck.

class TransactionSchema(pa.DataFrameModel):
    quantity: Series[int] = pa.Field(gt=0)
    unitprice: Series[float] = pa.Field(ge=0)


    @pa.check("unitprice")
    def priceprecision(cls, s: Series[float]) -> Series[bool]:

        return s.round(2).eq(s)

    @pa.dataframecheck
    def totalispositive(cls, df: pd.DataFrame) -> Series[bool]:
        return (df["quantity"]  df["unitprice"]) > 0

from pandera.typing import DataFrame


class RawTransactions(pa.DataFrameModel):
    transactionid: Series[int]

    quantity: Series[int]
    unitprice: Series[float]

    class Config:
        coerce = True


class EnrichedTransactions(RawTransactions):
    total: Series[float] = pa.Field(ge=0)


@pa.checktypes

def addtotal(df: DataFrame[RawTransactions]) -> DataFrame[EnrichedTransactions]:
    return df.assign(total=df["quantity"]  df["unitprice"])

Ketika addtotal dipanggil, Pandera memvalidasi argumen terhadap RawTransactions sebelum tubuh fungsi berjalan dan memvalidasi nilai kembalian terhadap EnrichedTransactions sesudahnya. Ini mengubah kesesuaian skema menjadi kontrak yang diperiksa di setiap batas fungsi, yang sangat berharga dalam rantai transformasi yang panjang.

Pemeriksaan yang Dapat Dipakai Ulang dan Terdaftar

Ketika logika kustom yang sama muncul di beberapa skema, daftarkan sekali dan rujuk dengan nama. Dekorator @pa.extensions.registercheckmethod menambahkan sebuah pemeriksaan ke namespace Check.

import pandera.extensions as extensions


@extensions.registercheckmethod(statistics=["maxdecimals"])

def hasmaxdecimals(pandasobj, , maxdecimals):

    factor = 10  maxdecimals
    return (pandasobj  factor).round() == (pandasobj  factor)



schema = DataFrameSchema(
    {
        "unitprice": Column(float, Check.hasmaxdecimals(maxdecimals=2)),
    }
)

import pandera as pa

bad = pd.DataFrame(
    {
        "transactionid": [1, 1, 3],          # id duplikat

        "product": ["Keyboard", "Mouse", "Monitor"],
        "category": ["Peripherals", "Unknown", "Displays"],  # kategori tidak valid
        "quantity": [2, -1, 1],               # quantity negatif
        "unitprice": [29.99, 14.50, 199.00],
        "region": ["West", "East", "West"],
        "customeremail": ["a@example.com", None, "c@example.com"],

    }
)

try:
    schema.validate(bad, lazy=True)
except pa.errors.SchemaErrors as exc:
    print(exc.failurecases)        # DataFrame berisi setiap kasus yang gagal
    print(exc.data)                 # data asli yang divalidasi

from pandera import Hypothesis

schema = DataFrameSchema(
    {
        "unitprice": Column(float),
        "region": Column(str),
    },
    checks=Hypothesis.twosamplettest(
        sample1="West",
        sample2="East",
        groupby="region",
        relationship="greaterthan",

        alpha=0.05,
    ),
)

from pandera import Index, MultiIndex

Index tunggal: index integer yang unik dan terurut
schema = DataFrameSchema(
    columns={"unitprice": Column(float)},
    index=Index(int, Check.greaterthanorequalto(0), unique=True),
)

MultiIndex: validasi setiap level
multi = DataFrameSchema(
    columns={"unitprice": Column(float)},

    index=MultiIndex(
        [
            Index(str, name="region"),
            Index(pd.Timestamp, name="date"),
        ]
    ),
)

from pandera.typing import Index, Series


class IndexedTransactions(pa.DataFrameModel):
    idx: Index[int] = pa.Field(ge=0, unique=True)
    unitprice: Series[float]

from pandera.typing import DataFrame


class RawSales(pa.DataFrameModel):
    transactionid: Series[int] = pa.Field(unique=True)
    product: Series[str]
    quantity: Series[int] = pa.Field(gt=0)
    unitprice: Series[float] = pa.Field(ge=0)


    class Config:
        coerce = True


class CleanSales(RawSales):
    revenue: Series[float] = pa.Field(ge=0)


class RegionSummary(pa.DataFrameModel):
    region: Series[str]
    totalrevenue: Series[float] = pa.Field(ge=0)


@pa.checktypes

def extract(path: str) -> DataFrame[RawSales]:
    return pd.readcsv(path)


@pa.checktypes

def transform(df: DataFrame[RawSales]) -> DataFrame[CleanSales]:
    return df.assign(revenue=df["quantity"]  df["unitprice"])


@pa.checktypes

def summarize(df: DataFrame[CleanSales]) -> DataFrame[RegionSummary]:
    return (
        df.groupby("region", asindex=False)["revenue"]
        .sum()
        .rename(columns={"revenue": "totalrevenue"})

    )

Jika extract membaca CSV dengan transactionid yang duplikat, kegagalan muncul seketika saat ekstraksi, bukan setelah data diagregasi dan baris aslinya hilang. Setiap skema juga mendokumentasikan apa yang diharapkan tahap tersebut, sehingga pipeline mendeskripsikan dirinya sendiri.

Memvalidasi DataFrame Polars dan PySpark

Pandera memakai satu model skema bersama lintas backend. Untuk Polars, impor kelas skema dari pandera.polars dan beri anotasi dengan modul typing Polars.

import polars as pl
import pandera.polars as pa
from pandera.typing.polars import Series


class PolarsTransactions(pa.DataFrameModel):
    transactionid: Series[int] = pa.Field(unique=True)

    quantity: Series[int] = pa.Field(gt=0)
    unitprice: Series[float] = pa.Field(ge=0)


df = pl.DataFrame(
    {"transactionid": [1, 2], "quantity": [1, 2], "unitprice": [9.9, 19.9]}
)
validated = PolarsTransactions.validate(df)

Dukungan PySpark mengikuti bentuk yang sama melalui pandera.pyspark. Kosakata pemeriksaan konsisten lintas mesin, sehingga skema yang Anda pahami untuk pandas terbaca sama untuk Polars. Perbedaan utamanya ada pada jalur impor dan kenyataan bahwa sebagian pemeriksaan khusus pandas mungkin tidak punya padanan di setiap backend.

Menyimpulkan dan Mengekspor Skema

Ketika Anda menghadapi dataset yang asing, biarkan Pandera menyusun rancangan skema dari sampel dengan pa.inferschema. Hasilnya adalah titik awal yang Anda perhalus dengan tangan; perlakukan batasan hasil inferensi sebagai saran, bukan aturan final.

import pandera as pa

inferred = pa.inferschema(df)
print(inferred)

Anda dapat menserialisasi skema ke skrip Python atau ke YAML untuk ditinjau dan dikontrol versinya.

# Tulis skrip Python yang dapat dijalankan untuk merekonstruksi skema
inferred.toscript("transactionschema.py")

Atau serialisasi ke YAML
inferred.toyaml("transactionschema.yaml")

Muat kembali skema dari YAML
schema = pa.DataFrameSchema.fromyaml("transactionschema.yaml")

Inferensi paling berguna sebagai langkah awal. Skema yang dihasilkan cenderung terlalu longgar di beberapa tempat dan terlalu ketat di tempat lain, jadi selalu tinjau sebelum mengandalkannya.

Praktik Terbaik

Letakkan skema berdekatan dengan kode yang menghasilkan data. Skema yang didefinisikan di samping fungsi transformasinya lebih mudah dijaga kebenarannya daripada yang berada di file konfigurasi yang jauh.

Gunakan DataFrameModel berbasis kelas untuk apa pun yang melampaui skrip cepat. Ia lebih mudah dibaca, mendukung pewarisan untuk skema yang berkaitan, dan bekerja dengan @pa.checktypes untuk memvalidasi batas fungsi.

Utamakan pemeriksaan vektorisasi dibanding elementwise demi performa, dan sediakan pemeriksaan hypothesis untuk pemantauan tingkat distribusi, bukan validasi per record.

Kontrol versi skema Anda bersama kode. Karena berupa Python biasa, skema layak berada di repositori dan proses tinjauan yang sama dengan pipeline yang dilindunginya.

Kesimpulan dan Poin Penting

Pandera membawa validasi data ke tempat yang sama dengan tempat data Anda ditransformasi: kode Python biasa. Dengan mendeklarasikan definisi DataFrameSchema atau DataFrameModel, melampirkan pemeriksaan bawaan dan kustom, serta memvalidasi pada batas fungsi dengan @pa.checktypes, Anda mengubah asumsi implisit tentang data menjadi kontrak eksplisit yang ditegakkan.

• Pandera ringan dan mengutamakan kode; pilih untuk validasi pipeline inline, dan pertimbangkan Great Expectations ketika Anda membutuhkan platform kualitas data bersama yang terdokumentasi.
• DataFrameSchema dan DataFrameModel berbasis kelas menyatakan batasan yang sama; API kelas lebih mudah diskalakan dan terintegrasi dengan type hint.
• Check bawaan mencakup aturan umum, pemeriksaan kustom menangani sisanya, dan elementwise menukar kecepatan dengan kejelasan per nilai.
• lazy=True mengumpulkan setiap kegagalan ke dalam objek SchemaErrors yang DataFrame failurecases-nya menunjukkan setiap masalah dengan tepat.
• Model skema yang sama memvalidasi pandas, Polars, dan PySpark, dan pa.inferschema ditambah ekspor YAML atau skrip membantu Anda memulai dan mengontrol versi skema.

Mulai dari yang kecil: tambahkan satu skema pada batas paling rapuh dalam pipeline Anda, validasi secara lazy untuk melihat seperti apa kenyataannya, lalu perketat aturan sampai skema menjadi deskripsi jujur dari data Anda.