用Pandas+Pydantic搭建数据洗濯与验证管道

数据洗濯是每个和数据打交道的人都绕不开的日常。
无论是做分析、跑模型，还是给业务方出报表，拿到的原始数据往往都带着重复记录、缺失值、格式错误，甚至一些显着违反常识的条目。
我以前习惯每次遇到问题就暂时写几行 drop_duplicates() 或 fillna()，但下次换一份数据，同样的脏法又得重写一遍。
与其如许反复救火，不如建一条可复用的数据管道：洗濯、验证、陈诉一步到位，让脏数据颠末流水线后变成可信的输入。
本文将用 Pandas 和 Pydantic 搭一条不到 100 行核心代码的数据洗濯与验证管道。
读完之后希望能让朋侪们明白管道的设计思路，还能把它直接套用到平常的生活和工作场景中。
1. 管道的三条职责：洗濯、验证、陈诉

制造业的流水线之所以高效，是因为每个工位只做一件事，上一步的输出就是下一步的输入。
数据管道也可以如许设计。

我们让管道承担三项职责：

• 洗濯，负责去重、处理缺失值，并为后续扩展留出空间；
  
• 验证，确保每一行数据都满意业务规则，比如年龄不能是负数、手机号必须是 11 位；
  
• 陈诉，追踪整个过程中到底去掉了多少重复、补了多少空、拦下了多少错误。
  

如许一来，数据质量不再是含糊的感觉，而是可以量化的指标。
要搭建这条管道，只必要两个库：pandas 用来处理表格，pydantic 用来定义“什么算有效数据”。
如果你还没安装，在终端里执行 pip install pandas pydantic 即可。
2. 先定义“有效”，再谈洗濯

很多人的洗濯习惯是先把数据改得“看起来能看”，再去想验证的事。
但更好的次序是反过来：先明确什么样的数据才算合格，然后再让洗濯服务于这个标准。
Pydantic 的 BaseModel 正好适合做这件事。
我们以一个社区团购订单为例，定义 OrderValidator，字段包括姓名、年龄、手机号和订单金额。

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

复制代码

这里有几个关键点：

• @field_validator 必须和 @classmethod 搭配使用；
  
• 年龄限制在 0 到 100 之间；
  
• 手机号要求 11 位数字且以 1 开头，这很符合国内手机号的格式；
  
• 金额不能为负。
  

有了这个模式，任何一行数据只要不满意规则，Pydantic 就会抛出 ValidationError，我们就能把它拦下来。
3. 给管道一个骨架，并让它学会记账

定义好验证模式后，我们创建一个 DataPipeline 类。
它的构造函数里初始化一个统计字典，用来记录去重数量、空值处理数量和验证错误数量。
别小看这个字典，它让管道有了“记账”的能力，每次运行后你都能清楚知道数据被动了哪些手脚。

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

复制代码

4. 洗濯：去重、补缺，但要有章法

洗濯方法 clean_data 吸收一个 DataFrame，返回处理后的版本。
第一步是去重，而且必须在填充缺失值之前做——如果先填充再取中位数，重复记录会把中位数拉偏。
去重之后，数值列用中位数填充，因为中位数比均值更抗离群值；文本列则统一填成 'Unknown'，给后续人工排查留个标记。
每处理一处空值，统计字典里的 nulls_handled 就累加一次。

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

复制代码

5. 验证：逐行检查，但不让坏数据中断管道

洗濯完成后，数据看起来完整了，但未必精确。
validate_data 方法会逐行迭代，把每一行转成字典后传给 OrderValidator。
有效行收集到 valid_rows，无效行则捕获 ValidationError，记录下行号和错误详情。
如许做的好处是，单条坏数据不会让整个管道瓦解——能处理的只管处理，同时把问题记录在案，供后续人工审查。

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

复制代码

6. 编排：让洗濯和验证串起来

最后，用一个 process 方法把洗濯和验证串联起来，返回一个包罗洗濯后数据、验证错误列表和统计信息的综合陈诉。
如许调用方只必要一个入口，就能拿到全部效果。

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

复制代码

7. 在社区团购订单场景中试一试

社区团购在国内非常普遍，订单数据通常来自微信群接龙、小程序下单或团长手动录入。
这类数据往往存在一些典范问题：同一用户重复下单、手机号少填几位、年龄栏被乱填、金额出现负数或空白。
我们构造一份示例数据来模拟这种情况：

1. if __name__ == "__main__":
2.     # 模拟社区团购订单数据：包含重复、缺失、非法年龄、非法手机号
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("\n清洗后的有效数据：")
28.     print(result["cleaned_data"])
30.     print("\n验证错误清单：")
31.     if result["validation_errors"]:
32.         for err in result["validation_errors"]:
33.             print(f"  行号 {err['row']}：{err['errors']}")
34.     else:
35.         print("  无")
37.     print("\n统计信息：")
38.     for key, value in result["stats"].items():
39.         print(f"  {key}: {value}")
41.     print("\n" + "=" * 60)
42.     print(
43.         "处理完成。有效数据 {} 行，错误 {} 行。".format(
44.             len(result["cleaned_data"]), len(result["validation_errors"])
45.         )
46.     )

复制代码

运行后，管道会先去掉一条完全重复的“李娜”记录，把缺失的姓名填成 'Unknown'，用中位数填补缺失的金额。
接着逐行验证：李娜的年龄 -5 和手机号 12345 会被拦下，刘洋的年龄 150 也会被标记为错误。
最终输出的有效数据里，张伟、王芳和那位姓名被补为 'Unknown' 的用户顺遂通过；错误清单则清晰列出问题所在的行和原因。

团长拿到这份陈诉，就能直接知道哪些订单必要人工核对手机号，哪些年龄显着是误填，而不必在一堆脏数据里反复翻找。
8. 管道之外：还能怎么扩展

这条管道只有不到 100 行核心代码，但它已经具备了可维护系统的雏形。
你可以根据详细场景继续扩展，比如：

• 增加自定义洗濯规则，比如标准化地址、脱敏手机号；
  
• 也可以让 Pydantic 模式可配置，使同一管道适配不同来源的数据；
  
• 如果数据量很大，还可以把逐行验证换成向量化操作或并行处理。
  

关键不在于一次写得多完美，而在于拥有一个可复用的骨架，把重复性的洗濯和验证工作交给管道，自己则把精神放在从干净数据中提取洞察上。

免责声明：如果侵犯了您的权益，请联系站长及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金.