Pandas、Polars 的神级助攻：Pandera
数
云朵君 云朵君约 9 分钟
云朵君 云朵君约 9 分钟01 先约定一张合格订单表应该是什么样
先说一下演示数据。这里用的
monthly_sales_dirty.csv
一共 1,200 行，日期从 2026 年 1 月 1 日到 10 月 20 日。另有一份
known_issues.csv
，记录了预先植入的 28 处问题：数量混入文字、区域缺失、单价为负、折扣大于 1、日期无效、订单号重复，以及渠道值不在已登记的范围内，每类各 4 处。
表里有
order_id
、
order_date
、
region
、
product
、
quantity
、
unit_price
、
discount_pct
和
channel
八列。在计算净销售额之前，至少要确认几件事：列不能缺；数量必须是正整数，单价必须大于零；折扣只能在 0 到 1 之间；订单号不能重复；区域、商品和渠道要符合已有的业务字典。
这些规则就是本例的数据契约。把它放在读取之后、分析之前，哪一列、哪一行不符合要求，可以尽早发现；等数据已经参与汇总，再找异常就难得多。不过 Pandera 只负责检查约定有没有被满足，至于错误记录该修复、拒收还是暂时隔离，还得由业务规则决定。
02 读 CSV 时保留原始值，再做类型解析
先安装依赖：python -m pip install pandas pandera plotly matplotlib
monthly_sales_dirty.csv
放到脚本同一目录。这里特意先把所有字段按字符串读取，并保留原始副本
raw
。这样能避免一开始就依赖 Pandas 的自动类型推断；空字符串也先原样留下，随后再按字段规则处理。
from pathlib import Path
import pandas as pd
DATA_PATH = Path("monthly_sales_dirty.csv")
raw = pd.read_csv(DATA_PATH, dtype="string", keep_default_na=False)
df = raw.copy()
# 保留原始行号，后续错误报告可以对应回 CSV 数据行（不含表头）。
df.index.name = "data_row"
required_columns = {
"order_id", "order_date", "region", "product",
"quantity", "unit_price", "discount_pct", "channel",
}
missing_columns = required_columns - set(df.columns)
if missing_columns:
raise ValueError(f"CSV 缺少必需列: {sorted(missing_columns)}")
for column in ["order_id", "order_date", "region", "product", "channel"]:
df[column] = df[column].str.strip().replace("", pd.NA)
df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce").astype("Int64")
df["unit_price"] = pd.to_numeric(df["unit_price"], errors="coerce")
df["discount_pct"] = pd.to_numeric(df["discount_pct"], errors="coerce")
df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
errors="coerce"
有个容易误会的地方：它只是把无法解析的值转成缺失值，并没有把错误修好。比如数量列里的文字会变成缺失值，交给后面的 schema 报告；要查原来填了什么，还可以回到
raw
找对应的单元格。
03 用 Pandera 声明列、范围和唯一性
数据类型解析好之后，就可以写 Pandera 的 schema 了。用
DataFrameSchema
一次声明列类型、是否允许缺失、唯一性、数值范围和合法类别。区域、商品、渠道的可选值在实际项目里应该来自业务字典；下面只是为了演示，把它们写成了固定集合。
import pandera.pandas as pa
from pandera import Check
schema = pa.DataFrameSchema(
{
"order_id": pa.Column(str, nullable=False, unique=True),
"order_date": pa.Column(pa.DateTime, nullable=False),
"region": pa.Column(
str, checks=Check.isin(["华东", "华南", "华北", "西南"]), nullable=False
),
"product": pa.Column(
str, checks=Check.isin(["鼠标", "键盘", "显示器", "扩展坞"]), nullable=False
),
"quantity": pa.Column(
pa.Int64, checks=[Check.gt(0)], nullable=False
),
"unit_price": pa.Column(float, checks=Check.gt(0), nullable=False),
"discount_pct": pa.Column(
float, checks=[Check.ge(0), Check.le(1)], nullable=False
),
"channel": pa.Column(
str, checks=Check.isin(["直营网店", "平台店", "线下门店"]), nullable=False
),
},
strict=True,
coerce=False,
)
strict=True
会把意外多出的列也视为契约不匹配；
coerce=False
则要求前面的解析步骤先把数据变成预期类型，而不是让 schema 默默替我们转型。若业务允许额外字段，可以按情况关闭严格列检查。类别值也不应照抄示例：当新渠道是正常业务变化时，应先更新字典和契约，而不是把新值一概认成脏数据。
04 一次找出失败项，再隔离异常记录
校验时可以先用
lazy=True
：让 Pandera 把能发现的失败项一次收集起来，而不是报出第一处问题就停。这样更容易看出一批数据到底坏在哪些地方。下面采取一个简单的处理策略：只要订单的关键字段没有通过校验，就先隔离整行，不让它参与本次分析。
from pandera.errors import SchemaErrors
try:
clean = schema.validate(df, lazy=True)
quarantine = df.iloc[0:0].copy()
failure_cases = pd.DataFrame()
except SchemaErrors as exc:
failure_cases = exc.failure_cases
# 先保存完整报告；没有行号的失败项不能简单按行隔离。
failure_cases.to_csv("validation-failures.csv", index=False)
if failure_cases["index"].isna().any():
raise ValueError(
"存在无法按行隔离的 schema 错误，请先检查列或表级约束"
) from exc
row_indexes = failure_cases["index"].unique()
quarantine = df.loc[df.index.isin(row_indexes)].copy()
clean = df.drop(index=row_indexes).copy()
# 隔离后再校验一次，确保下游只拿到符合契约的数据。
clean = schema.validate(clean, lazy=True)
quarantine.to_csv("quarantine.csv", index=True)
print(f"通过的行数: {len(clean)}")
print(f"隔离的行数: {len(quarantine)}")
print(failure_cases.head(10))
failure_cases
里的行索引隔离，并对留下的
clean
再做一次校验。如果出现缺列等无法定位到具体行的错误，代码会中止，而不是把其余记录当成合格数据继续画图。
真正接入流水线时，
validation-failures.csv
和
quarantine.csv
最好连同批次号、来源文件、校验时间一起保存，方便回查。本例采取的是“关键字段有问题就整行隔离”的办法；如果业务允许修正单个字段，或者同一行其他字段仍有用途，就需要另外制定处理策略。
别把“植入了多少处问题”和“隔离了多少行”混为一谈。清单标记了 28 处异常，但订单号唯一性检查会把重复键涉及的两条记录都找出来；同一行也可能触发不止一条检查。最终数量要以失败报告为准。
要看问题集中在哪些字段，可以把
failure_cases
按列汇总。在这份样本中，订单号唯一性对应 8 条失败记录，其余六类字段各 4 条。
import plotly.express as px
failure_summary = (
failure_cases.groupby("column", as_index=False)
.size()
.rename(columns={"size": "failure_cases"})
)
rule_names = {
"order_id": "订单号唯一性",
"order_date": "日期有效性",
"region": "区域必填",
"quantity": "数量类型/范围",
"unit_price": "单价范围",
"discount_pct": "折扣范围",
"channel": "渠道字典",
}
failure_summary["rule"] = failure_summary["column"].map(rule_names)
fig = px.bar(
failure_summary,
x="failure_cases",
y="rule",
orientation="h",
text="failure_cases",
title="Pandera 校验失败项：按字段统计",
labels={"failure_cases": "失败记录数", "rule": "校验字段"},
)
fig.show()
exc.message
；
exc.failure_cases
则是便于筛选、导出的表格形式。字段细节以当前 Pandera Pandas 错误报告文档为准，换到其他后端时不要默认结构完全相同。
05 校验过关，再把数据交给图表
现在
clean
已经过隔离后的二次校验，可以用它来计算净销售额：
quantity × unit_price × (1 - discount_pct)
。先按月份、区域汇总，再按时间排序后绘制折线图。因为用的是合成订单，图标题也明确标出“样本”，避免被当成真实经营数据。
import plotly.express as px
chart_data = clean.assign(
month=clean["order_date"].dt.to_period("M").dt.to_timestamp(),
net_sales=clean["quantity"].astype(float)
* clean["unit_price"]
* (1 - clean["discount_pct"]),
)
monthly = (
chart_data.groupby(["month", "region"], as_index=False)["net_sales"]
.sum()
.sort_values(["month", "region"])
)
fig = px.line(
monthly,
x="month",
y="net_sales",
color="region",
markers=True,
title="合成订单样本：各区域月度净销售额",
labels={"month": "月份", "net_sales": "净销售额（样本金额）", "region": "区域"},
)
fig.show()
color
为区域分组；折线图的横轴顺序取决于传入数据的顺序，因此先排序能避免月份连线错乱。图只说明“如何把通过规则的数据接到可视化”，不是现实销售趋势，也不能证明校验规则本身符合业务口径。
chart_data
，换一个分组维度即可，不用为每张图重复写一套校验逻辑：
product_sales = (
chart_data.groupby("product", as_index=False)["net_sales"]
.sum()
.sort_values("net_sales", ascending=False)
)
fig = px.bar(
product_sales,
x="product",
y="net_sales",
text_auto=".2s",
title="通过校验的样本净销售额：按商品汇总",
labels={"product": "商品", "net_sales": "净销售额（样本金额）"},
)
fig.show()outcome = pd.DataFrame({
"状态": ["通过校验", "隔离"],
"记录数": [len(clean), len(quarantine)],
})
fig = px.pie(
outcome,
names="状态",
values="记录数",
hole=0.6,
title="本批次校验结果：按记录数统计",
)
fig.show()06 换成 Polars，这套校验能直接搬过去吗？
Pandera 不只支持 Pandas，也提供了 Polars 集成。不过，有集成不等于把上面的
pandera.pandas
schema 原样复制过去就能跑。两个后端的支持范围、执行方式和错误报告都有可能不同。
如果项目准备迁移到 Polars，先对照官方后端支持表，检查现有规则在目标版本里是否可用，再用项目样本验证懒执行、异常收集和错误处理。尤其是依赖
failure_cases
行索引来隔离记录的逻辑，不应未经验证直接复用。
07 写在最后
什么时候值得引入 Pandera？如果一张 DataFrame 会反复进入报表、模型或团队共享的处理流程，而且列、类型、唯一性、取值范围这些要求已经比较稳定，Pandera 就很适合放在数据读取后的入口位置。规则集中写一次，后面的分析步骤只消费通过校验的数据，也更方便排查是哪一批输入出了问题。 但它不是每个小脚本都必须引入的依赖。一次性的探索分析，或者业务规则还没定下来时，先用简单断言和抽样检查可能更省事。真要把 Pandera 接进生产流程，别只写 schema：业务字典、库与后端版本、异常记录的保存和处理责任也得一起约定清楚。 Pandera 能回答的是“这份数据是否符合我们写下的规则”。至于规则写得对不对，依旧要回到业务本身。
参考材料
- Pandera 稳定版文档与后端支持矩阵 — pandera.readthedocs.io
- Pandera Pandas 错误报告 — pandera.readthedocs.io
- Pandera Polars 集成说明 — pandera.readthedocs.io
- Plotly Python 折线图文档 — plotly.com