当我们处理数据时，确保数据不脏、不无效非常重要——比如检查空值、缺失值，或某列类型不允许的数字。这些检查至关重要，因为劣质数据会导致错误分析、模型失败，并浪费大量时间和资源。
你可能已经用传统的 Pandas 方法清理和验证过数据，但在本教程中，我想向你介绍一款更强大的 Python 库：Pandera。Pandera 提供了灵活且表达力强的 API，可对类 DataFrame 对象进行数据验证。与手动检查相比，它更快、更具扩展性。你只需创建定义数据结构、数据类型和规则的 schema（模式），然后 Pandera 会根据这些 schema 检查你的数据，指出不符合要求的地方，让你能在早期发现并修正问题，避免日后遇到大麻烦。
本指南假设你已经具备一定的 Python 和 Pandas 基础。让我们一步步学习如何在工作流中使用 Pandera。
第 1 步：环境配置
首先，需要安装必要的包：
pip install pandera pandas
安装完成后，导入所需库，并验证安装结果：
import pandas as pd
import pandera as paprint("pandas version:", pd.__version__)
print("pandera version:", pa.__version__)
输出应如下，显示 pandas 和 Pandera 的版本，确认安装成功：
pandas version: 2.2.2
pandera version: 0.0.0+dev0
第 2 步：创建示例数据集
让我们创建一个包含有意错误的客户信息数据集，便于演示清洗和验证流程：
import pandas as pd# 包含错误的客户数据集
data = pd.DataFrame({"customer_id": [1, 2, 3, 4, "invalid"], # "invalid" 不是整数"name": ["Maryam", "Jane", "", "Alice", "Bobby"], # 包含空名字"age": [25, -5, 30, 45, 35], # 年龄为负数无效"email": ["mrym@gmail.com", "jane.s@yahoo.com", "invalid_email", "alice@google.com", None] # 邮箱无效或缺失
})print("原始 DataFrame：")
print(data)
输出：
原始 DataFrame：customer_id name age email
0 1 Maryam 25 mrym@gmail.com
1 2 Jane -5 jane.s@yahoo.com
2 3 30 invalid_email
3 4 Alice 45 alice@google.com
4 invalid Bobby 35 None
数据集中的问题：
customer_id：包含字符串 "invalid"，应为整数
name：存在空字符串
age：包含负数（-5）
email：格式无效（invalid_email）和缺失值（None）
第 3 步：定义 Pandera Schema
Pandera 的 schema 定义了 DataFrame 期望的结构和约束。我们将用 DataFrameSchema 为每一列指定规则：
import pandera as pa
from pandera import Column, Check, DataFrameSchema# 定义 schema
schema = DataFrameSchema({"customer_id": Column(dtype="int64",checks=[Check.isin(range(1, 1000)), # ID 在 1~999 之间Check(lambda x: x > 0, element_wise=True) # 必须为正数],nullable=False),"name": Column(dtype="string",checks=[Check.str_length(min_value=1),Check(lambda x: x.strip() != "", element_wise=True)],nullable=False),"age": Column(dtype="int64",checks=[Check.greater_than(0), # 必须为正Check.less_than_or_equal_to(120) # 合理范围],nullable=False),"email": Column(dtype="string",checks=[Check.str_matches(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")],nullable=False)
})
第 4 步：初始验证
现在，用 schema 验证 DataFrame。Pandera 的 validate 方法会检查数据是否符合 schema，设置 lazy=True 可收集所有错误：
print("\n初始验证：")
try:validated_df = schema.validate(data, lazy=True)print("数据有效！")print(validated_df)
except pa.errors.SchemaErrors as e:print("验证失败，具体问题如下：")print(e.failure_cases[['column', 'check', 'failure_case', 'index']])
验证会失败，错误信息类似：
验证失败，具体问题如下：column check failure_case index
0 customer_id isin(range(1, 1000)) invalid 4
1 name str_length(1, None) 2
2 name 2
3 age greater_than(0) -5 1
4 email not_nullable None 4
5 email str_matches('...') invalid_email 2
6 customer_id dtype('int64') object None
...
第 5 步：清洗数据
我们已发现问题，接下来逐步清洗数据：
customer_id：移除非整数或无效 ID
name：移除空字符串
age：移除负数或不合理年龄
email：移除无效或缺失邮箱
# 5a. 清洗 customer_id
data["customer_id"] = pd.to_numeric(data["customer_id"], errors="coerce")
data = data[data["customer_id"].notna()]
data = data[data["customer_id"].isin(range(1, 1000))]
data["customer_id"] = data["customer_id"].astype("int64")# 5b. 清洗 name
data = data[data["name"].str.strip() != ""]
data["name"] = data["name"].astype("string[python]")# 5c. 清洗 age
data = data[data["age"] > 0]
data = data[data["age"] <= 120]# 5d. 清洗 email
data = data[data["email"].notna()]
data = data[data["email"].str.match(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")]
data["email"] = data["email"].astype("string[python]")print("清洗后的 DataFrame：")
print(data)
清洗后 DataFrame：
清洗后的 DataFrame：customer_id name age email
0 1 Maryam 25 mrym@gmail.com
3 4 Alice 45 alice@google.com
第 6 步：再次验证数据
验证清洗后的 DataFrame，确保其符合 schema：
print("\n最终验证：")
try:validated_df = schema.validate(data, lazy=True)print("清洗后的数据有效！")print(validated_df)
except pa.errors.SchemaErrors as e:print("清洗后验证失败，错误如下：")print(e.failure_cases[['column', 'check', 'failure_case', 'index']])
输出：
最终验证：
清洗后的数据有效！customer_id name age email
0 1 Maryam 25 mrym@gmail.com
3 4 Alice 45 alice@google.com
第 7 步：构建可复用的数据处理管道
可以将清洗和验证流程封装为可复用的函数：
def process_data(df, schema):"""用 Pandera schema 处理并验证 DataFrame。参数：df: 输入的 pandas DataFrameschema: Pandera DataFrameSchema返回：验证且清洗后的 DataFrame，若验证失败则返回 None"""data_clean = df.copy()data_clean["customer_id"] = pd.to_numeric(data_clean["customer_id"], errors="coerce")data_clean = data_clean[data_clean["customer_id"].notna()]data_clean = data_clean[data_clean["customer_id"].isin(range(1, 1000))]data_clean["customer_id"] = data_clean["customer_id"].astype("int64")data_clean = data_clean[data_clean["name"].str.strip() != ""]data_clean["name"] = data_clean["name"].astype("string")data_clean = data_clean[data_clean["age"] > 0]data_clean = data_clean[data_clean["age"] <= 120]data_clean["age"] = data_clean["age"].astype("int64")data_clean = data_clean[data_clean["email"].notna()]data_clean = data_clean[data_clean["email"].str.match(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")]data_clean["email"] = data_clean["email"].astype("string")data_clean = data_clean.reset_index(drop=True)try:validated_df = schema.validate(data_clean, lazy=True)print("数据处理成功！")return validated_dfexcept pa.errors.SchemaErrors as e:print("清洗后验证失败，错误如下：")print(e.failure_cases[['column', 'check', 'failure_case', 'index']])return None# 测试管道
print("\n测试管道：")
final_df = process_data(data, schema)
print("最终处理后的 DataFrame：")
print(final_df)
输出：
测试管道：
数据处理成功！
最终处理后的 DataFrame：customer_id name age email
0 1 Maryam 25 mrym@gmail.com
1 4 Alice 45 alice@google.com
结论
Pandera 是保障 pandas 数据工作流数据质量的强大工具。通过定义 schema，你能及早发现错误，强制执行数据一致性，并自动化数据清洗。本文你学到了：
-
安装 Pandera 并设置样例数据集
-
定义包含类型和约束规则的 schema
-
验证数据并定位问题
-
清洗数据以符合 schema
-
再次验证清洗后的数据
-
构建可复用的数据处理管道
Pandera 还支持更复杂的验证场景，例如基于类的 schema、跨字段校验、局部校验等，详情可参见官方 Pandera 文档。
