Pandas结合Pydantic构建数据清洗与验证管道

脚本专家

数据清洗是数据分析和脚本开发中绕不开的环节。原始表里常见重复记录、缺失值、格式错误，甚至明显违背业务常识的条目。临时写 drop_duplicates() 或 fillna() 能救急，但换一份数据又要重写。更稳妥的方式是用 Pandas 处理表格，用 Pydantic 定义有效规则，再把清洗、验证、报告串成一条可复用的数据管道。

依赖安装很简单：

2. pip install pandas pydantic

复制代码

一、先定义什么算有效数据

Pydantic 的 BaseModel 适合描述业务规则。以社区团购订单为例，字段包括姓名、年龄、手机号和订单金额。年龄限制在 0 到 100 之间，手机号必须是 11 位数字且以 1 开头，金额不能为负数。注意 @field_validator 需要和 @classmethod 搭配使用。

2. from pydantic import BaseModel, field_validator, ValidationError
3. from typing import Optional
4. import pandas as pd
5. import numpy as np
7. class OrderValidator(BaseModel):
8.     name: str
9.     age: Optional[int] = None
10.     phone: Optional[str] = None
11.     amount: Optional[float] = None
13.     @field_validator('age')
14.     @classmethod
15.     def validate_age(cls, v):
16.         if v is not None and (v < 0 or v > 100):
17.             raise ValueError('年龄必须在0到100之间')
18.         return v
20.     @field_validator('phone')
21.     @classmethod
22.     def validate_phone(cls, v):
23.         if v and (not v.isdigit() or len(v) != 11 or not v.startswith('1')):
24.             raise ValueError('手机号必须是11位数字且以1开头')
25.         return v
27.     @field_validator('amount')
28.     @classmethod
29.     def validate_amount(cls, v):
30.         if v is not None and v < 0:
31.             raise ValueError('金额不能为负数')
32.         return v

复制代码

二、管道骨架与统计记账

定义好验证模式后，创建 DataPipeline。构造函数中初始化 cleaning_stats，记录 duplicates_removed、nulls_handled 和 validation_errors。这样每次运行后都能知道去掉了多少重复、补了多少空、拦下了多少错误。

2. class DataPipeline:
3.     def __init__(self):
4.         self.cleaning_stats = {
5.             'duplicates_removed': 0,
6.             'nulls_handled': 0,
7.             'validation_errors': 0
8.         }

复制代码

三、清洗：先去重，再补缺

clean_data 接收 DataFrame，返回处理后的版本。第一步是 drop_duplicates()，而且必须在填充缺失值之前执行。原因是重复记录会影响中位数计算，如果先填充再取中位数，统计基准会被重复行拉偏。去重之后，数值列用中位数填充，因为中位数比均值更抗离群值；文本列统一填成 Unknown，方便后续人工排查。每处理一处空值，nulls_handled 就累加对应数量。

2. def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
3.         initial_rows = len(df)
4.         df = df.drop_duplicates()
5.         self.cleaning_stats['duplicates_removed'] = initial_rows - len(df)
7.         numeric_columns = df.select_dtypes(include=[np.number]).columns
8.         for col in numeric_columns:
9.             null_count = df[col].isnull().sum()
10.             if null_count > 0:
11.                 df[col] = df[col].fillna(df[col].median())
12.                 self.cleaning_stats['nulls_handled'] += null_count
14.         string_columns = df.select_dtypes(include=['object']).columns
15.         for col in string_columns:
16.             null_count = df[col].isnull().sum()
17.             if null_count > 0:
18.                 df[col] = df[col].fillna('Unknown')
19.                 self.cleaning_stats['nulls_handled'] += null_count
21.         return df

复制代码

四、验证：逐行检查，坏数据不中断管道

清洗后的数据看起来完整，但不一定正确。validate_data 会逐行迭代，把每一行转成字典后传给 OrderValidator。有效行收集到 valid_rows，无效行捕获 ValidationError，并记录行号和错误详情。这样单条坏数据不会让整个管道崩溃，能处理的尽量处理，问题则记录在案供人工审查。

2. def validate_data(self, df: pd.DataFrame):
3.         valid_rows = []
4.         errors = []
5.         for idx, row in df.iterrows():
6.             try:
7.                 validated = OrderValidator(**row.to_dict())
8.                 valid_rows.append(validated.model_dump())
9.             except ValidationError as e:
10.                 errors.append({'row': idx, 'errors': str(e)})
11.         self.cleaning_stats['validation_errors'] = len(errors)
12.         return pd.DataFrame(valid_rows), errors

复制代码

五、编排：把清洗和验证串起来

process 方法作为统一入口，先用 df.copy() 调用 clean_data，再把清洗结果交给 validate_data，最后返回 cleaned_data、validation_errors 和 stats 的综合报告。调用方只需要一个入口就能拿到全部结果。

2. def process(self, df: pd.DataFrame):
3.         cleaned_df = self.clean_data(df.copy())
4.         validated_df, validation_errors = self.validate_data(cleaned_df)
5.         return {
6.             'cleaned_data': validated_df,
7.             'validation_errors': validation_errors,
8.             'stats': self.cleaning_stats
9.         }

复制代码

六、社区团购订单场景实测

社区团购订单常来自微信群接龙、小程序下单或团长手动录入，容易出现同一用户重复下单、手机号少填、年龄乱填、金额为负或空白等情况。下面构造一份示例数据：

2. if __name__ == '__main__':
3.     sample_data = pd.DataFrame(
4.         {
5.             'name': ['张伟', '李娜', '王芳', None, '刘洋', '李娜'],
6.             'age': [28, -5, 35, 42, 150, -5],
7.             'phone': [
8.                 '13812345678',
9.                 '12345',
10.                 '13987654321',
11.                 '13712345678',
12.                 '13612345678',
13.                 '12345',
14.             ],
15.             'amount': [45.5, 60.0, 45.5, None, 88.8, 60.0],
16.         }
17.     )
19.     print('=' * 60)
20.     print('原始数据：')
21.     print(sample_data)
22.     print('=' * 60)
24.     pipeline = DataPipeline()
25.     result = pipeline.process(sample_data)
27.     print()
28.     print('清洗后的有效数据：')
29.     print(result['cleaned_data'])
30.     print()
31.     print('验证错误清单：')
32.     if result['validation_errors']:
33.         for err in result['validation_errors']:
34.             print(' 行号 {}：{}'.format(err['row'], err['errors']))
35.     else:
36.         print(' 无')
37.     print()
38.     print('统计信息：')
39.     for key, value in result['stats'].items():
40.         print(' {}: {}'.format(key, value))
41.     print()
42.     print('=' * 60)
43.     print(
44.         '处理完成。有效数据 {} 行，错误 {} 行。'.format(
45.             len(result['cleaned_data']), len(result['validation_errors'])
46.         )
47.     )

复制代码

运行后，管道会先去掉一条完全重复的“李娜”记录，把缺失的姓名填成 Unknown，用中位数填补缺失的金额。接着逐行验证：李娜的年龄 -5 和手机号 12345 会被拦下，刘洋的年龄 150 也会被标记为错误。最终有效数据中，张伟、王芳以及姓名被补为 Unknown 的用户通过；错误清单列出问题行号和原因。团长拿到报告后，可以直接定位哪些订单需要核对手机号，哪些年龄明显是误填。

七、可扩展方向

这条管道核心代码不到 100 行，但已经具备可维护系统的雏形。可以增加自定义清洗规则，例如标准化地址、脱敏手机号；也可以让 Pydantic 模式可配置，使同一管道适配不同来源的数据；数据量很大时，还能把逐行验证换成向量化操作或并行处理。重点不是一次写得多完美，而是拥有可复用骨架，把重复清洗和验证交给管道，把精力放在从干净数据中提取洞察上。

通过 Pandas 负责表格操作、Pydantic 负责规则校验、DataPipeline 负责编排和统计，一条可复用的数据清洗与验证管道就完成了。

热心网友3

感谢分享，这个把清洗、验证、统计串成管道的思路很清晰，复用性确实比临时 drop_duplicates 和 fillna 强很多。不过我照着代码看了一遍，发现 clean_data 里数值列和文本列填充那段可能有点小问题：`null_count = df.isnull().sum()` 返回的是整表的 Series，不是当前列的数量；而且 `df.fillna(df.median())` 会一次性对所有列填充，文本列也可能被中位数填进去，容易出问题。建议改成 `null_count = df.isnull().sum()`，然后按列填充，比如 `df = df.fillna(df.median())`，文本列同理用 `df.fillna('Unknown')`。 另外验证环节，文本列缺失已经被填成了 `'Unknown'`，手机号验证会把它当成非数字直接拦截，导致所有原本缺失的手机号都变成无效记录。可能需要在验证时把 `'Unknown'` 当作跳过项，或者在清洗阶段用更明确的缺失标记，再在验证模型里区分处理。还有逐行 iterrows 在数据量大时确实偏慢，可以考虑用 `apply` 或批量转换后交给 Pydantic 验证，性能会好一些。 整体框架很实用，统计记账这点也很赞，期待后续把验证部分补完。

热心网友6

这个组合挺清晰，Pandas 做表格清洗、Pydantic 做规则约束，再用统计字段把去重、补空、验证错误数量记下来，复用性和可观测性都照顾到了。先去重再填中位数、文本缺失填 Unknown 这些选择也合理，尤其避免重复行影响中位数这点很实在。 看代码时有两个地方可能需要再顺一下：数值列循环里 `null_count = df.isnull().sum()` 拿到的是整表每列空值数，不是当前列的空值数，后面判断和累加容易出问题，建议直接用当前列来取。填充时 `df.fillna(df.median())` 也会把各列的中位数混着填，最好按列填充，文本列也一样。另外字段类型可以写明确一些，比如 Optional、Optional、Optional，验证器里对手机号做 isdigit 前也要确认拿到的是字符串。整体骨架很好，等后面把验证结果报告和错误明细补全就更完整了。

热心网友5

这个管道思路很清晰，Pydantic 管规则、Pandas 管表格，再配上清洗统计，后续复盘会方便很多。先 drop_duplicates 再 fillna 这个顺序确实关键，不然中位数容易被重复行带偏。 有两个小地方可以留意：一是 null_count = df.isnull().sum() 拿到的是每列空值数的 Series，后面直接 if null_count > 0 在部分 pandas 版本里容易报“真值不明确”，最好按列统计；二是手机号如果被读成整数，验证时 isdigit 可能先报错，清洗阶段或模型里统一转成字符串会更稳。 逐行验证、收集错误、不让坏数据中断管道这点很实用。后面 validate_data 没贴完，期待补全错误明细和报告输出部分。