用 Pandera 高效验证和清洗 Pandas 数据集——实用分步指南

当我们处理数据时，确保数据不脏、不无效非常重要——比如检查空值、缺失值，或某列类型不允许的数字。这些检查至关重要，因为劣质数据会导致错误分析、模型失败，并浪费大量时间和资源。

你可能已经用传统的 Pandas 方法清理和验证过数据，但在本教程中，我想向你介绍一款更强大的 Python 库：Pandera。Pandera 提供了灵活且表达力强的 API，可对类 DataFrame 对象进行数据验证。与手动检查相比，它更快、更具扩展性。你只需创建定义数据结构、数据类型和规则的 schema（模式），然后 Pandera 会根据这些 schema 检查你的数据，指出不符合要求的地方，让你能在早期发现并修正问题，避免日后遇到大麻烦。

本指南假设你已经具备一定的 Python 和 Pandas 基础。让我们一步步学习如何在工作流中使用 Pandera。

第 1 步：环境配置

首先，需要安装必要的包：

pip install pandera pandas

安装完成后，导入所需库，并验证安装结果：

import pandas as pd
import pandera as pa

print("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-.]+#34;)
        ],
        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-.]+#34;)]
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 DataFrame
        schema: 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-.]+#34;)]
    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_df
    except 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 文档。