{
"cells": [
{
"cell_type": "markdown",
"id": "0",
"metadata": {},
"source": [
"---\n",
"title: \"Chapter 26: Data schema validation with Pandera\"\n",
"---"
]
},
{
"cell_type": "markdown",
"id": "1",
"metadata": {},
"source": [
"[](https://colab.research.google.com/github/sambaiga/ds-mlops-path/blob/main/tutorials/02-dev-tools/26-pandera-schema-validation.ipynb) [](https://raw.githubusercontent.com/sambaiga/ds-mlops-path/main/tutorials/02-dev-tools/26-pandera-schema-validation.ipynb)"
]
},
{
"cell_type": "markdown",
"id": "2",
"metadata": {},
"source": [
"The grade-predictor pipeline finishes training. Accuracy: 87%. You deploy it.\n",
"\n",
"Three weeks later a data engineer flags something: 23 rows in `university_analytics.csv` have `final_score = 999`. The data entry system used 999 as a sentinel for \"not recorded\". Nobody documented it. The model trained on those 23 rows as if 999 were a valid exam score. There was no error, no warning. The model just learned the wrong thing for 23 students.\n",
"\n",
"The check you ran at load time called `.dtypes` and printed `.head()`. Neither caught a score of 999. Validation by inspection runs once, in a notebook cell, by whoever is at the keyboard. It is forgotten the next time the pipeline runs.\n",
"\n",
"What you needed was a **data contract**: a formal specification of what the DataFrame must look like: column types, allowed value ranges, uniqueness constraints, that runs automatically every time data enters the pipeline and fails loudly with a precise error when violated. Pydantic gave you this for individual records at the system boundary. Pandera gives you this for whole DataFrames inside the pipeline.\n",
"\n",
"**Next:** [Chapter 27: Logging](27-logging.ipynb) adds the audit trail that records which validation errors occurred and when, the permanent record that Pandera's exceptions do not keep by themselves.\n",
"\n",
"[](https://github.com/sambaiga/ds-mlops-path/blob/main/tutorials/02-dev-tools/26-pandera-schema-validation.ipynb)"
]
},
{
"cell_type": "markdown",
"id": "3",
"metadata": {},
"source": [
"::: {.callout-note collapse=\"true\" icon=false}\n",
"## Learning objectives\n",
"\n",
"By the end of this chapter you will be able to:\n",
"\n",
"| # | Skill | Covered in |\n",
"| --- | --- | --- |\n",
"| 1 | Explain the difference between row-level (Pydantic) and schema-level (Pandera) validation | Sec. 1 |\n",
"| 2 | Define a Pandera `DataFrameSchema` with column types and constraints | Sec. 2 |\n",
"| 3 | Use the class-based API with `pa.DataFrameModel` for typed schemas | Sec. 3 |\n",
"| 4 | Write custom element-wise and series-level checks | Sec. 4 |\n",
"| 5 | Validate DataFrames in a pipeline and collect errors without stopping | Sec. 5 |\n",
"| 6 | Use Pandera schemas as pytest fixtures to document data contracts | Sec. 6 |\n",
":::"
]
},
{
"cell_type": "markdown",
"id": "4",
"metadata": {},
"source": [
"Pydantic validated the shape of a single record as it entered the system: one row, one pass or fail. A DataFrame is not a record. It is a table. The properties that matter at the table level are different: is `student_id` unique across all 2,400 rows? Is the distribution of `program` values consistent with what the source system should produce? Does `final_score` stay between 0 and 100 for every row, not just the one you checked manually?\n",
"\n",
"These are questions Pydantic cannot answer, because Pydantic operates on one object at a time.\n",
"\n",
"| | Pydantic `BaseModel` | Pandera `DataFrameSchema` |\n",
"| --- | --- | --- |\n",
"| Unit | One record (row) | Whole DataFrame |\n",
"| Checks | Type coercion, field constraints, cross-field | Column dtype, nullability, uniqueness, value ranges, statistical bounds |\n",
"| Returns | Validated model instance | Validated DataFrame |\n",
"| Error info | Field path + message | Row index + column + failed check |\n",
"| Best for | API inputs, config objects | CSVs, pipeline data, feature tables |\n",
"\n",
"\n",
" Key concept: a schema is documentation that runs
\n",
"A Pandera schema is not a test you write once. It is a living specification: the schema says what the DataFrame must look like, and the pipeline is only allowed to proceed when the data keeps that promise. Every time the pipeline runs, the contract is checked. When the data changes in a way that violates it, you know immediately, with a row index, a column name, and the rule that failed.\n",
"
"
]
},
{
"cell_type": "markdown",
"id": "5",
"metadata": {},
"source": [
"The diagram below shows the contract in action: a raw DataFrame enters Pandera, the schema checks each column's type, nullability, range, and uniqueness, and either returns a validated frame or raises a `SchemaError` with the exact row index and rule that failed.\n",
"\n",
"{fig-alt=\"Three columns: raw DataFrame with highlighted bad rows (None, 102, empty string); Pandera schema in the centre listing column rules; validated frame on the right containing only conforming rows. A red SchemaError arrow exits downward from the schema when data fails.\"}"
]
},
{
"cell_type": "markdown",
"id": "6",
"metadata": {},
"source": [
"## 1. Defining a `DataFrameSchema`\n",
"\n",
"The functional API creates a schema by passing a dict of column names to `pa.Column` definitions. Each `pa.Column` declares the dtype and the checks it must satisfy."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "7",
"metadata": {},
"outputs": [],
"source": [
"import numpy as np\n",
"import pandas as pd\n",
"import pandera as pa"
]
},
{
"cell_type": "markdown",
"id": "8",
"metadata": {},
"source": [
"Set `coerce=True` to let Pandera convert `\"78.5\"` to `78.5` before checking a `float` column. This mirrors Pydantic's type coercion at the row level. The top-level `checks` list accepts table-level constraints such as uniqueness across the entire column."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9",
"metadata": {},
"outputs": [],
"source": [
"schema = pa.DataFrameSchema(\n",
" {\n",
" \"student_id\": pa.Column(str, pa.Check.str_matches(r\"^S\\d{4}$\")),\n",
" \"midterm_score\": pa.Column(float, [pa.Check.ge(0.0), pa.Check.le(100.0)]),\n",
" \"final_score\": pa.Column(float, [pa.Check.ge(0.0), pa.Check.le(100.0)]),\n",
" \"project_score\": pa.Column(float, [pa.Check.ge(0.0), pa.Check.le(100.0)]),\n",
" \"program\": pa.Column(str, pa.Check.isin([\"CS\", \"EE\", \"Math\", \"Physics\", \"Biology\"])),\n",
" \"has_internet\": pa.Column(bool),\n",
" \"school_id\": pa.Column(str, nullable=True),\n",
" \"teacher_count\": pa.Column(int, pa.Check.ge(1)),\n",
" \"school_size\": pa.Column(str, pa.Check.isin([\"Small\", \"Medium\", \"Large\"])),\n",
" \"pass_threshold\": pa.Column(float, pa.Check.between(50.0, 80.0), nullable=True),\n",
" },\n",
" checks=[\n",
" pa.Check(lambda df: df[\"student_id\"].is_unique, error=\"student_id must be unique\"),\n",
" ],\n",
" coerce=True,\n",
")\n",
"\n",
"print(\"Schema created with\", len(schema.columns), \"columns\")"
]
},
{
"cell_type": "markdown",
"id": "10",
"metadata": {},
"source": [
"Build a small valid sample to test against:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "11",
"metadata": {},
"outputs": [],
"source": [
"sample_df = pd.DataFrame(\n",
" {\n",
" \"student_id\": [\"S0001\", \"S0002\", \"S0003\"],\n",
" \"midterm_score\": [78.5, 65.0, 90.0],\n",
" \"final_score\": [82.0, 70.0, 88.0],\n",
" \"project_score\": [91.0, 75.0, 85.0],\n",
" \"program\": [\"CS\", \"EE\", \"Math\"],\n",
" \"has_internet\": [True, False, True],\n",
" \"school_id\": [\"SCH01\", \"SCH01\", \"SCH02\"],\n",
" \"teacher_count\": [12, 8, 15],\n",
" \"school_size\": [\"Large\", \"Medium\", \"Large\"],\n",
" \"pass_threshold\": [60.0, 60.0, 65.0],\n",
" }\n",
")\n",
"sample_df.head()"
]
},
{
"cell_type": "markdown",
"id": "12",
"metadata": {},
"source": [
"Call `.validate()` on a conforming DataFrame. It returns the validated frame unchanged:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "13",
"metadata": {},
"outputs": [],
"source": [
"validated = schema.validate(sample_df)\n",
"print(\"Validation passed:\", validated.shape)"
]
},
{
"cell_type": "markdown",
"id": "14",
"metadata": {},
"source": [
"Introduce a row with `midterm_score = 150` and confirm Pandera raises `SchemaError` with the row index and the violated rule:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "15",
"metadata": {},
"outputs": [],
"source": [
"bad_df = sample_df.copy()\n",
"bad_df.loc[0, \"midterm_score\"] = 150.0\n",
"\n",
"try:\n",
" schema.validate(bad_df)\n",
"except pa.errors.SchemaError as e:\n",
" print(type(e).__name__)\n",
" print(e)"
]
},
{
"cell_type": "markdown",
"id": "16",
"metadata": {},
"source": [
"\n",
"
Activity 1: duplicate student_id\n",
"Create a DataFrame where two rows share the same
student_id. Run
schema.validate(df) and confirm it raises a
SchemaError with a message about uniqueness.\n",
"
dup_df = sample_df.copy()\n",
"dup_df.loc[1, \"student_id\"] = \"S0001\" # duplicate\n",
"schema.validate(dup_df) # should raise\n",
"
\n",
" Pro Tip: Use DataFrameModel for reuse, DataFrameSchema for quick scripts
\n",
"DataFrameModel is easier to subclass, document, and test: it reads like a dataclass and fits naturally alongside Pydantic models. DataFrameSchema is useful when you want to build a schema programmatically at runtime, e.g., from a config file or database metadata.\n",
"
\n",
"\n",
"\n",
"
Activity 2: extend the Schema\n",
"Subclass
StudentDataSchema and add a
semester column constrained to
[\"Fall\", \"Spring\", \"Summer\"]. Validate a DataFrame that includes the column with valid values, then one that has an invalid value. Confirm only the second raises a
SchemaError.\n",
"
class ExtendedSchema(StudentDataSchema):\n",
" semester: Series[str] = pa.Field(isin=[\"Fall\", \"Spring\", \"Summer\"])\n",
"
\n",
"
Activity 3: cross-Column Check\n",
"Add a
@pa.dataframe_check to
GradeSchema that verifies
midterm_score and
final_score are not both 0 for the same student (a student with both scores at 0 is almost certainly an error, not a result). Confirm it passes on valid data and fails when you introduce a row with both at 0.\n",
"
@pa.dataframe_check\n",
"def not_both_zero(cls, df: pd.DataFrame) -> pd.Series:\n",
" return ~((df[\"midterm_score\"] == 0) & (df[\"final_score\"] == 0))\n",
"
\n",
" Key Concept: lazy=True collects all errors; lazy=False stops on the first
\n",
"schema.validate(df, lazy=False) (the default) raises a SchemaError on the first failure: fast and clear for development. schema.validate(df, lazy=True) collects every failure and raises a SchemaErrors (note the plural) at the end, better for production, where you want a full error report rather than a partial run.\n",
"
\n",
"\n",
"\n",
" Activity 4: full Error Report
\n",
"Introduce three invalid rows into sample_df: one with midterm_score=150.0, one with an invalid program, and one with a duplicate student_id. Call StudentDataSchema.validate(bad_df, lazy=True). Catch the SchemaErrors exception and print the failure_cases DataFrame showing all three failures at once.\n",
"
"
]
},
{
"cell_type": "markdown",
"id": "39",
"metadata": {},
"source": [
"## 5. Schemas as data contracts in tests\n",
"\n",
"A Pandera schema is documentation that runs. Put it in a pytest fixture and every test that touches a DataFrame automatically validates the contract."
]
},
{
"cell_type": "markdown",
"id": "40",
"metadata": {},
"source": [
"Define `StudentDataSchema` in `grade_predictor/schemas.py` so it can be imported by both the pipeline and the tests:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "41",
"metadata": {},
"outputs": [],
"source": [
"# grade_predictor/schemas.py\n",
"import pandera as pa\n",
"from pandera.typing import Series\n",
"\n",
"\n",
"class StudentDataSchema(pa.DataFrameModel):\n",
" student_id: Series[str] = pa.Field(str_matches=r\"^S\\d{4}$\")\n",
" midterm_score: Series[float] = pa.Field(ge=0.0, le=100.0)\n",
" final_score: Series[float] = pa.Field(ge=0.0, le=100.0)\n",
" project_score: Series[float] = pa.Field(ge=0.0, le=100.0)\n",
" program: Series[str] = pa.Field(isin=[\"CS\", \"EE\", \"Math\", \"Physics\", \"Biology\"])\n",
" has_internet: Series[bool]\n",
" teacher_count: Series[int] = pa.Field(ge=1)\n",
" school_size: Series[str] = pa.Field(isin=[\"Small\", \"Medium\", \"Large\"])\n",
"\n",
" class Config:\n",
" coerce = True"
]
},
{
"cell_type": "markdown",
"id": "42",
"metadata": {},
"source": [
"A `@pytest.fixture` builds a minimal valid DataFrame. Each test gets a fresh copy:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "43",
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd\n",
"import pytest\n",
"\n",
"\n",
"@pytest.fixture\n",
"def valid_df():\n",
" return pd.DataFrame(\n",
" {\n",
" \"student_id\": [\"S0001\", \"S0002\"],\n",
" \"midterm_score\": [78.5, 65.0],\n",
" \"final_score\": [82.0, 70.0],\n",
" \"project_score\": [91.0, 75.0],\n",
" \"program\": [\"CS\", \"EE\"],\n",
" \"has_internet\": [True, False],\n",
" \"teacher_count\": [12, 8],\n",
" \"school_size\": [\"Large\", \"Medium\"],\n",
" }\n",
" )"
]
},
{
"cell_type": "markdown",
"id": "44",
"metadata": {},
"source": [
"Test that clean data passes the schema:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "45",
"metadata": {},
"outputs": [],
"source": [
"def test_valid_data_passes_schema(valid_df):\n",
" validated = StudentDataSchema.validate(valid_df)\n",
" assert len(validated) == 2 # noqa: S101"
]
},
{
"cell_type": "markdown",
"id": "46",
"metadata": {},
"source": [
"Test that a score above 100 raises `SchemaError`:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "47",
"metadata": {},
"outputs": [],
"source": [
"def test_invalid_score_raises(valid_df):\n",
" bad = valid_df.copy()\n",
" bad.loc[0, \"midterm_score\"] = 150.0\n",
" with pytest.raises(pa.errors.SchemaError):\n",
" StudentDataSchema.validate(bad)"
]
},
{
"cell_type": "markdown",
"id": "48",
"metadata": {},
"source": [
"Test that an unknown program value raises `SchemaError`:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "49",
"metadata": {},
"outputs": [],
"source": [
"def test_invalid_program_raises(valid_df):\n",
" bad = valid_df.copy()\n",
" bad.loc[0, \"program\"] = \"Underwater Basket Weaving\"\n",
" with pytest.raises(pa.errors.SchemaError):\n",
" StudentDataSchema.validate(bad)"
]
},
{
"cell_type": "markdown",
"id": "50",
"metadata": {},
"source": [
"Run the tests from the repo root:"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "51",
"metadata": {},
"outputs": [],
"source": [
"# Save the cells above to tests/test_data_contracts.py, then:\n",
"print(\"Run with: uv run pytest tests/test_data_contracts.py -v\")"
]
},
{
"cell_type": "markdown",
"id": "52",
"metadata": {},
"source": [
"\n",
" Pro Tip: Use pa.DataFrameModel.example() to generate test data automatically
\n",
"Pandera can synthesise valid DataFrames from a schema: StudentDataSchema.example(size=50) generates 50 valid rows matching all constraints. This removes the need to hand-craft test fixtures for every new schema.\n",
"
\n",
"\n",
"\n",
"
Activity 5: schema-Driven Test Data\n",
"Call
StudentDataSchema.example(size=20) to generate 20 synthetic rows. Confirm that the generated DataFrame passes
StudentDataSchema.validate() without errors. Then confirm that if you corrupt one cell (set a score to 200), validation fails.\n",
"
synthetic = StudentDataSchema.example(size=20)\n",
"StudentDataSchema.validate(synthetic) # should pass\n",
"
\n",
"
Capstone - End-to-End Validated Pipeline\n",
"
\n",
"- Define
StudentDataSchema in grade_predictor/schemas.py covering all columns of university_analytics.csv \n",
"- Update
load_and_validate (from Chapter 25) to run Pandera schema validation after row-level Pydantic validation \n",
"- Add a
@pa.dataframe_check that verifies the computed weighted average (using the weights from PipelineConfig) falls in [0, 100] for every row \n",
"- Write three tests: one that the real CSV passes the schema, one that a DataFrame with a bad score raises
SchemaError, and one that uses example() to generate synthetic data and confirms it passes \n",
"- Run
uv run pytest -v and confirm all three pass \n",
"
\n",
"
\n",
" Key Concept: Schema validation is not the same as data cleaning
\n",
"Schema validation answers a yes/no question: does this DataFrame conform to the contract? It runs fast and fails loudly. Data cleaning transforms and repairs data. The two serve different purposes: validate at the pipeline boundary, clean before you get there.\n",
"
\n",
"\n",
"| Tool | Best for | Limitation |\n",
"| --- | --- | --- |\n",
"| **Pandera** | DataFrame contracts in Python pipelines | pandas/polars-specific; not for arbitrary dicts |\n",
"| **Great Expectations** | Enterprise data quality suites, HTML reports, data docs | Heavy setup, complex configuration, overkill for most ML work |\n",
"| **Pydantic** | Row-level validation, API payloads, config models | Not designed for tabular data; looping over rows is slow |\n",
"| **Frictionless Data** | YAML-defined schemas, cross-language contracts | Less Python-native; validation logic lives in config files |\n",
"| **Cerberus / marshmallow** | Dict and object validation | No DataFrame support; designed for records not tables |\n",
"\n",
"The practical split comes down to scope. Pydantic is the right choice when you are validating individual records at a system boundary: an API payload, a config file, a single row entering a pipeline. Pandera is the right choice when you are validating the structure and statistics of an entire DataFrame. They are complementary, not competing: Chapter 25 showed you Pydantic for the row, this notebook shows you Pandera for the table.\n",
"\n",
"Great Expectations is powerful but designed for data engineering teams who need data documentation, alerting, and audit trails across multiple data sources. For most ML projects, Pandera gives you 90% of the value with 10% of the setup.\n",
"\n",
"\n",
" Pro Tip: Use both Pydantic and Pandera in the same pipeline
\n",
"Validate incoming JSON records with Pydantic as they arrive, then validate the assembled DataFrame with Pandera before it enters any model or transformation. The two checks run at different granularities and catch different classes of bugs: Pydantic catches a bad individual record, Pandera catches drift in the distribution across hundreds of records.\n",
"
"
]
},
{
"cell_type": "markdown",
"id": "55",
"metadata": {},
"source": [
"## Further reading\n",
"\n",
"| Resource | Why it matters |\n",
"| --- | --- |\n",
"| [Pandera documentation](https://pandera.readthedocs.io/) | Full reference: DataFrameSchema, DataFrameModel, check types, Polars support |\n",
"| [Pandera + Polars](https://pandera.readthedocs.io/en/stable/polars.html) | Same API works on Polars DataFrames |\n",
"| [Pandera pytest integration](https://pandera.readthedocs.io/en/stable/pandas_on_spark.html) | `@pa.check_types` decorator for function-level schema enforcement |\n",
"| [Pydantic + Pandera together](https://pandera.readthedocs.io/en/stable/pydantic_integration.html) | Use a Pandera schema as a Pydantic field type |\n",
"| [Great Expectations](https://docs.greatexpectations.io/) | Enterprise-grade data quality platform; Pandera for code-first, GX for teams with data catalogs |"
]
},
{
"cell_type": "markdown",
"id": "56",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"| Concept | Key rule |\n",
"| --- | --- |\n",
"| `DataFrameSchema` | Functional API: describe columns, constraints, table-level checks as a dict |\n",
"| `DataFrameModel` | Class-based API: columns as annotated fields, checks as classmethods |\n",
"| `pa.Field(ge=, le=)` | Same constraint vocabulary as Pydantic's `Field` |\n",
"| `pa.Check.isin([...])` | Restrict a column to an explicit set of values |\n",
"| `@pa.check(\"col\")` | Column-level custom check: receives the Series, returns bool or bool Series |\n",
"| `@pa.dataframe_check` | Cross-column check: receives the full DataFrame |\n",
"| `lazy=True` | Collect all failures; raises `SchemaErrors` (plural) |\n",
"| `lazy=False` | Stop on first failure; raises `SchemaError` |\n",
"| `schema.example(size=n)` | Generate n rows of valid synthetic data for testing |\n",
"| Pydantic + Pandera | Validate rows with Pydantic, validate the assembled table with Pandera |\n",
"\n",
"**Next:** Chapter 27: Classical ML: Scikit-learn pipelines applied to the fully validated, typed `grade-predictor` dataset."
]
}
],
"metadata": {
"kernelspec": {
"display_name": "ark",
"language": "python",
"name": "ark"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.12"
}
},
"nbformat": 4,
"nbformat_minor": 5
}