How to Build Production-Grade Data Validation Pipelines Using Pandera, Typed Schemas, and Composable DataFrame Contracts
Schemas, and Composable DataFrame contractsIn this tutorial, we show how to build robust production-grade data validation pipelines using Pandera with typed DataFrame models. We start by simulating real, incomplete transaction data and gradually enforce strict schema constraints, column-level rules, and cross-column business logic using validation checks. We show how lazy validation helps us identify multiple data quality issues at once, how invalid records can be closed without breaking pipelines, and how schema enforcement can be applied directly to functional boundaries to ensure correctness as data flows through the transition. Check it out FULL CODES here.
!pip -q install "pandera>=0.18" pandas numpy polars pyarrow hypothesis
import json
import numpy as np
import pandas as pd
import pandera as pa
from pandera.errors import SchemaError, SchemaErrors
from pandera.typing import Series, DataFrame
print("pandera version:", pa.__version__)
print("pandas version:", pd.__version__)Set up the workstation by installing Pandera and its dependencies and importing all required libraries. We verify library versions to ensure reproducibility and compatibility. It establishes a clean basis for enforcing the validation of typed data throughout the course. Check it out FULL CODES here.
rng = np.random.default_rng(42)
def make_raw_orders(n=250):
countries = np.array(["CA", "US", "MX"])
channels = np.array(["web", "mobile", "partner"])
raw = pd.DataFrame(
{
"order_id": rng.integers(1, 120, size=n),
"customer_id": rng.integers(1, 90, size=n),
"email": rng.choice(
["[email protected]", "[email protected]", "bad_email", None],
size=n,
p=[0.45, 0.45, 0.07, 0.03],
),
"country": rng.choice(countries, size=n, p=[0.5, 0.45, 0.05]),
"channel": rng.choice(channels, size=n, p=[0.55, 0.35, 0.10]),
"items": rng.integers(0, 8, size=n),
"unit_price": rng.normal(loc=35, scale=20, size=n),
"discount": rng.choice([0.0, 0.05, 0.10, 0.20, 0.50], size=n, p=[0.55, 0.15, 0.15, 0.12, 0.03]),
"ordered_at": pd.to_datetime("2025-01-01") + pd.to_timedelta(rng.integers(0, 120, size=n), unit="D"),
}
)
raw.loc[rng.choice(n, size=8, replace=False), "unit_price"] = -abs(raw["unit_price"].iloc[0])
raw.loc[rng.choice(n, size=6, replace=False), "items"] = 0
raw.loc[rng.choice(n, size=5, replace=False), "discount"] = 0.9
raw.loc[rng.choice(n, size=4, replace=False), "country"] = "ZZ"
raw.loc[rng.choice(n, size=3, replace=False), "channel"] = "unknown"
raw.loc[rng.choice(n, size=6, replace=False), "unit_price"] = raw["unit_price"].iloc[:6].round(2).astype(str).values
return raw
raw_orders = make_raw_orders(250)
display(raw_orders.head(10))We create a virtual data set that intentionally covers common data quality issues. We simulate invalid values, inconsistent types, and unexpected categories to reflect real-world import scenarios. It allows us to logically test and demonstrate the effectiveness of schema-based authentication. Check it out FULL CODES here.
EMAIL_RE = r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+.[A-Za-z]{2,}$"
class Orders(pa.DataFrameModel):
order_id: Series[int] = pa.Field(ge=1)
customer_id: Series[int] = pa.Field(ge=1)
email: Series[object] = pa.Field(nullable=True)
country: Series[str] = pa.Field(isin=["CA", "US", "MX"])
channel: Series[str] = pa.Field(isin=["web", "mobile", "partner"])
items: Series[int] = pa.Field(ge=1, le=50)
unit_price: Series[float] = pa.Field(gt=0)
discount: Series[float] = pa.Field(ge=0.0, le=0.8)
ordered_at: Series[pd.Timestamp]
class Config:
coerce = True
strict = True
ordered = False
@pa.check("email")
def email_valid(cls, s: pd.Series) -> pd.Series:
return s.isna() | s.astype(str).str.match(EMAIL_RE)
@pa.dataframe_check
def total_value_reasonable(cls, df: pd.DataFrame) -> pd.Series:
total = df["items"] * df["unit_price"] * (1.0 - df["discount"])
return total.between(0.01, 5000.0)
@pa.dataframe_check
def channel_country_rule(cls, df: pd.DataFrame) -> pd.Series:
ok = ~((df["channel"] == "partner") & (df["country"] == "MX"))
return okWe describe a robust Pandera DataFrameModel that captures both structural and business-level constraints. We use column-level rules, regex-based validation, and extensive data frame checks to declaratively enforce domain sanity. Check it out FULL CODES here.
try:
validated = Orders.validate(raw_orders, lazy=True)
print(validated.dtypes)
except SchemaErrors as exc:
display(exc.failure_cases.head(25))
err_json = exc.failure_cases.to_dict(orient="records")
print(json.dumps(err_json[:5], indent=2, default=str))We validate the raw dataset using lazy testing to reveal multiple violations in a single pass. We examine the schema failure scenarios to understand exactly where and why the data violates the schema rules. It helps us fix data quality issues without disrupting the entire pipeline. Check it out FULL CODES here.
def split_clean_quarantine(df: pd.DataFrame):
try:
clean = Orders.validate(df, lazy=False)
return clean, df.iloc[0:0].copy()
except SchemaError:
pass
try:
Orders.validate(df, lazy=True)
return df.copy(), df.iloc[0:0].copy()
except SchemaErrors as exc:
bad_idx = sorted(set(exc.failure_cases["index"].dropna().astype(int).tolist()))
quarantine = df.loc[bad_idx].copy()
clean = df.drop(index=bad_idx).copy()
return Orders.validate(clean, lazy=False), quarantine
clean_orders, quarantine_orders = split_clean_quarantine(raw_orders)
display(quarantine_orders.head(10))
display(clean_orders.head(10))
@pa.check_types
def enrich_orders(df: DataFrame[Orders]) -> DataFrame[Orders]:
out = df.copy()
out["unit_price"] = out["unit_price"].round(2)
out["discount"] = out["discount"].round(2)
return out
enriched = enrich_orders(clean_orders)
display(enriched.head(5))We separate valid records from invalid by separating rows that fail the schema test. We then apply schema guarantees to operational boundaries to ensure that only trusted data is changed. This pattern enables safe data enrichment while preventing silent corruption. Check it out FULL CODES here.
class EnrichedOrders(Orders):
total_value: Series[float] = pa.Field(gt=0)
class Config:
coerce = True
strict = True
@pa.dataframe_check
def totals_consistent(cls, df: pd.DataFrame) -> pd.Series:
total = df["items"] * df["unit_price"] * (1.0 - df["discount"])
return (df["total_value"] - total).abs() <= 1e-6
@pa.check_types
def add_totals(df: DataFrame[Orders]) -> DataFrame[EnrichedOrders]:
out = df.copy()
out["total_value"] = out["items"] * out["unit_price"] * (1.0 - out["discount"])
return EnrichedOrders.validate(out, lazy=False)
enriched2 = add_totals(clean_orders)
display(enriched2.head(5))We extend the basic schema with a derived column and ensure cross-column consistency using composable schemas. We ensure that the computed values obey the strict numerical values after conversion. It shows how Pandera supports the safety aspect of engineering with compelling guarantees.
In conclusion, we have developed a straightforward data validation method that treats schemes as first-class contracts rather than optional safeguards. We showed how schema architecture enables us to safely extend data sets with derived features while preserving variables, and how Pandera integrates seamlessly into data analysis and creation workflows. Through this tutorial, we’ve ensured that every transformation works on trusted data, allowing us to build pipelines that are transparent, debuggable, and resilient to real-world scenarios.
Check it out FULL CODES here. Also, feel free to follow us Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to Our newspaper. Wait! are you on telegram? now you can join us on telegram too.
