Pandas 数据清洗 进阶篇

上篇是基础清洗：缺失、重复、异常、日期、文本。本篇聚焦工业实战场景：复杂脏文本正则清洗、空字符串/空白字符区分、数据校验框架 Pandera、批量质量报告、高级异常识别、链式清洗流水线、脏数据分流、脱敏、内存优化。
前置依赖

import pandas as pd
import numpy as np
import re
from pandera import DataFrameSchema, Column, Check

一、区分：NaN / None / 空字符串 / 全空白字符串

基础清洗很容易踩坑：isnull() 识别不了 ""、" "（空格、tab、换行）。

df = pd.DataFrame({
    "备注": ["发货", "", None, "   ", "\t", np.nan, "破损"]
})

# 1. 判断是否是空字符串
df["is_empty_str"] = df["备注"] == ""
# 2. 判断是否全是空白（空格、制表符、换行）
df["is_whitespace"] = df["备注"].str.strip().eq("")
# 3. 统一标记脏空值（NaN + 空串 + 全空白）
df["is_bad_empty"] = df["备注"].isna() | df["备注"].str.strip().eq("")

统一替换方案：把空白类脏值统一转为 NaN，后续用 fillna / dropna 统一处理

def blank_to_nan(s: pd.Series):
    return s.where(s.str.strip().ne(""), np.nan)

df["备注"] = blank_to_nan(df["备注"])

二、正则表达式复杂文本清洗（高频实战）

适合：混杂符号、金额、手机号、编号、括号备注、多余换行、特殊字符。

示例数据集

df = pd.DataFrame({
    "手机号": ["138-0000-1111", "139 2222 3333", "号码未知", "137【4444】5555", "12345"],
    "金额": ["¥1,234.50", "2345元", "3000", "无效", "¥99.9"],
    "商品": ["键盘(有线)", "鼠标【无线】", "显示器-27寸_黑色"]
})

2.1 提取数字（去掉所有非数字字符）

# 提取手机号数字
df["phone_clean"] = df["手机号"].str.extract(r"(\d+)")
# 只保留11位手机号
df["phone_valid"] = df["phone_clean"].str.extract(r"1[3-9]\d{9}")

2.2 清洗金额，去除货币符号、千分位逗号

df["amount_clean"] = df["金额"].str.replace(r"[¥,元]", "", regex=True)
# 转数值，无法解析转为 NaN
df["amount"] = pd.to_numeric(df["amount_clean"], errors="coerce")

2.3 提取括号内文字 / 清除括号标记

# 清除【】() 以及里面内容
df["商品_simple"] = df["商品"].str.replace(r"[(\[【].*?[)\]】]", "", regex=True).str.strip()
# 单独提取括号内属性
df["attr"] = df["商品"].str.extract(r"[(\[【](.*?)[)\]】]")

2.4 通用正则清洗封装函数

def regex_clean_text(s: pd.Series, pattern: str, repl=""):
    return s.str.replace(pattern, repl, regex=True).str.strip()

三、高级异常值识别（不止IQR）

上篇只讲了IQR四分位，工业场景多种异常策略：

1. 业务阈值校验（优先，业务规则 > 统计规则）
2. 3σ 标准差法（适合正态分布数据）
3. 分位数截断（固定上下限百分位）
4. 标记异常，分流，而不是直接删除

df = pd.DataFrame({"销售额": np.random.normal(1000, 200, size=100)})
# 3σ 识别异常
mean = df["销售额"].mean()
std = df["销售额"].std()
upper = mean + 3 * std
lower = mean - 3 * std
df["is_outlier_3sigma"] = ~df["销售额"].between(lower, upper)

# 分位数截断：保留 0.5% ~ 99.5%
q_low = df["销售额"].quantile(0.005)
q_high = df["销售额"].quantile(0.995)
df["is_outlier_qcut"] = ~df["销售额"].between(q_low, q_high)

最佳实践：不要直接删除异常，分成两个数据集：正常数据 + 异常脏数据单独落地保存，用于复盘数据源问题

df_normal = df[~df["is_outlier_3sigma"]]
df_error = df[df["is_outlier_3sigma"]]

四、Pandera 结构化数据校验（强推荐，生产环境）

Pandera 给 DataFrame 定义 Schema，自动校验：列名、类型、取值范围、唯一约束、正则、自定义检查，提前拦截脏数据。
安装：pip install pandera

import pandera as pa
from pandera.typing import Series

# 定义校验模型
class SalesSchema(pa.SchemaModel):
    订单号: Series[str] = pa.Field(
        str_matches=r"^A\d{3}$",  # 正则匹配 A001,A002
        nullable=False
    )
    数量: Series[float] = pa.Field(ge=1, le=100)  # 大于等于1，小于等于100
    单价: Series[float] = pa.Field(gt=0)
    客户: Series[str] = pa.Field(nullable=True)

    class Config:
        strict = True  # 不允许出现schema以外的列

# 使用
schema = SalesSchema.to_schema()
# 校验，出错抛出异常
# df_valid = schema.validate(df)

# 校验并且捕获错误，不中断程序
validate_result = schema.validate(df, lazy=True)

• lazy=True：收集全部错误一次性返回，而不是遇到第一条脏数据就中断
• 适合定时ETL脚本，自动生成校验失败报告

五、自动生成数据质量报告（ydata-profiling）

一键生成HTML质量报告：缺失率、重复、分布、相关性、异常值
安装：pip install ydata-profiling

from ydata_profiling import ProfileReport

profile = ProfileReport(df, title="销售明细质量报告", explorative=True)
profile.to_file("data_quality.html")

适合：数据交付、对接业务、溯源脏数据。

六、链式清洗流水线（可复用、可调试）

使用 pipe 串联清洗函数，清晰、方便开关每个清洗步骤，替代嵌套函数。

def step_blank_to_nan(df):
    df["客户"] = df["客户"].where(df["客户"].str.strip().ne(""), np.nan)
    return df

def step_parse_date(df):
    df["下单日期"] = pd.to_datetime(df["下单日期"], errors="coerce")
    return df

def step_filter_business(df):
    df = df[(df["数量"]>0) & (df["单价"]>0)]
    return df

def step_mark_outlier(df):
    q1 = df["单价"].quantile(0.25)
    q3 = df["单价"].quantile(0.75)
    iqr = q3 - q1
    df["单价异常"] = ~df["单价"].between(q1 - 1.5*iqr, q3 + 1.5*iqr)
    return df

# 链式调用
df_clean = (
    df_raw
    .pipe(step_blank_to_nan)
    .pipe(step_parse_date)
    .pipe(step_filter_business)
    .pipe(step_mark_outlier)
)

优点：可以单独注释某一个pipe，关闭对应清洗规则，方便调试不同清洗策略。

七、字符串脱敏、敏感信息清洗

实战场景：手机号、身份证、客户姓名脱敏

# 手机号脱敏：138****1111
def mask_phone(s: pd.Series):
    return s.str.replace(r"(1\d{3})\d{4}(\d{4})", r"\1****\2", regex=True)

df["手机号_mask"] = mask_phone(df["手机号_valid"])

八、分类脏值映射 + 模糊匹配（字典映射 + 模糊合并）

业务经常出现同类名称多种写法：罗技鼠标 / 罗技-鼠标 / 罗技无线鼠标
简单替换不够，进阶：fuzzywuzzy 模糊匹配合并
安装：pip install fuzzywuzzy python-Levenshtein

from fuzzywuzzy import process

std_name = ["罗技鼠标", "戴尔键盘", "三星显示器"]
def fuzzy_match(name):
    if pd.isna(name):
        return None
    match, score = process.extractOne(name, std_name)
    return match if score >= 80 else "未匹配"

df["商品标准名"] = df["商品"].apply(fuzzy_match)

适合：爬虫、多渠道导入、人工录入的脏分类数据。

九、清洗后内存优化（大数据集清洗配套）

清洗完之后，优化类型，减少内存占用，适合十万、百万级行

# 分类列转category
df["商品名称"] = df["商品名称"].astype("category")
# 数值降级
df["数量"] = pd.to_numeric(df["数量"], downcast="integer")
df["单价"] = pd.to_numeric(df["单价"], downcast="float")

十、脏数据落地归档方案（生产ETL标准流程）

原始数据 df_raw
    ↓
统一空白转NaN
    ↓
字段类型转换、日期解析
    ↓
Pandera Schema校验
    ↓
拆分：正常数据 / 校验失败脏数据
    ↓
脏数据单独写入 error.xlsx，打上错误标签
    ↓
正常数据继续聚合、入库

示例分流代码

try:
    df_pass = schema.validate(df, lazy=True)
except pa.errors.SchemaErrors as err:
    err_df = err.failure_cases
    err_df.to_excel("data_error.xlsx", index=False)

十一、进阶避坑清单

1. 正则默认 regex=True，简单字符替换用 regex=False 速度更快
2. str.extract 没有匹配到返回NaN，不会报错
3. Pandera lazy模式适合离线ETL；严格模式适合实时校验，直接阻断
4. fuzzywuzzy 大数据很慢，百万行不要直接apply，建议先分组再匹配
5. 清洗流程一定要保留原始数据，所有清洗基于副本，禁止原地修改原始df
6. 区分：业务异常（脏数据） 和 正常极端业务数据，不要盲目统计剔除