数据清洗与预处理

数据分析圈有一句老话：数据清洗占了数据分析工作量的 80%。这不是夸张。在实际项目中，你拿到的原始数据几乎总是有问题的——缺失值、异常值、重复记录、格式不统一、噪声数据……如果不在分析之前把这些坑填平，后面的建模和可视化都会建立在沙地上。

本文是进阶篇的第一篇，我们会系统地讲解数据清洗与预处理的完整流程，每步都配有可运行的 Pandas 和 scikit-learn 代码。

数据质量维度

在动手清洗之前，先建立一套评估数据质量的框架。通常从四个维度衡量：

维度	含义	典型问题
准确性	数据是否真实反映客观事实	录入错误、单位混淆
完整性	数据是否有缺失	空值、Null、NaN
一致性	数据格式和单位是否统一	日期格式不一、大小写混用
时效性	数据是否在有效时间范围内	过期数据、时区偏差

理解这些维度能帮你快速定位问题类型，而不是盲目地套用清洗函数。

环境准备

本文所有代码基于以下依赖：

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.preprocessing import LabelEncoder, OneHotEncoder, StandardScaler

# 显示设置
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', 20)

我们用一个模拟的客户数据集来演示：

np.random.seed(42)
n = 1000

df = pd.DataFrame({
    'id': range(n),
    'age': np.random.randint(18, 70, n).astype(float),
    'income': np.random.normal(15000, 5000, n).astype(float),
    'gender': np.random.choice(['M', 'F', None], n, p=[0.45, 0.45, 0.1]),
    'city': np.random.choice(['北京', '上海', '广州', '深圳', None], n, p=[0.3, 0.25, 0.2, 0.15, 0.1]),
    'signup_date': pd.date_range('2020-01-01', periods=n, freq='D'),
    'score': np.random.uniform(0, 100, n),
    'phone': ['138' + str(np.random.randint(10000000, 99999999)) for _ in range(n)]
})

# 故意制造一些脏数据
df.loc[10:30, 'age'] = np.nan
df.loc[50:70, 'income'] = np.nan
df.loc[100:120, 'score'] = np.nan
df.loc[200, 'age'] = 200       # 异常值
df.loc[201, 'age'] = -5        # 异常值
df.loc[202, 'income'] = 999999 # 异常值
df.loc[300:305] = df.loc[300:305]  # 重复行
df.loc[400, 'phone'] = '138-1234-5678'  # 格式不一致
df.loc[401, 'phone'] = '13812345678'

缺失值处理

检测缺失值

首先摸清缺失情况：

# 查看每列缺失值数量
print(df.isnull().sum())

# 缺失比例
print(df.isnull().mean().round(4) * 100)

# 可视化缺失值
import missingno as msno
msno.matrix(df)
plt.show()

缺失比例决定了处理策略：低于 5% 可直接删除，5%-20% 需要填充，超过 20% 就要考虑这个特征是否还能用。

删除缺失值

# 删除包含任何缺失值的行
df_dropped = df.dropna()

# 删除某列有缺失的行
df_dropped = df.dropna(subset=['age'])

# 删除缺失比例超过 50% 的列
df_clean = df.dropna(thresh=len(df) * 0.5, axis=1)

直接删除简单粗暴，但如果缺失数据有某种规律，删除可能导致偏差。比如高收入客户更倾向于不填收入——删除它们会让收入分布失真。

均值 / 中位数 / 众数填充

# 均值填充（适合正态分布）
df['age_filled_mean'] = df['age'].fillna(df['age'].mean())

# 中位数填充（适合有偏分布）
df['income_filled_median'] = df['income'].fillna(df['income'].median())

# 众数填充（适合分类变量）
df['city_filled'] = df['city'].fillna(df['city'].mode()[0])

选择哪种统计量取决于数据分布。偏态严重的用中位数，有离群点的用中位数，对称分布用均值。

前向填充与后向填充

对于时间序列数据，相邻值通常更有参考价值：

# 前向填充——用上一个有效值填充
df['score_ffill'] = df['score'].fillna(method='ffill')

# 后向填充——用下一个有效值填充
df['score_bfill'] = df['score'].fillna(method='bfill')

# 限制最大填充步数（最多向前填充 2 步）
df['score_limited'] = df['score'].fillna(method='ffill', limit=2)

插值法

插值比简单填充更精确，它利用已知点的趋势来估计缺失点：

# 线性插值
df['score_interp'] = df['score'].interpolate(method='linear')

# 时间插值（适用于等间隔时间序列）
df['score_time'] = df['score'].interpolate(method='time')

# 多项式插值
df['score_poly'] = df['score'].interpolate(method='polynomial', order=2)

scikit-learn 中的缺失值处理

from sklearn.impute import SimpleImputer

# 均值填充
imputer = SimpleImputer(strategy='mean')
age_imputed = imputer.fit_transform(df[['age']])

# 中位数填充
imputer_med = SimpleImputer(strategy='median')
income_imputed = imputer_med.fit_transform(df[['income']])

# 最频繁值（众数）填充
imputer_mode = SimpleImputer(strategy='most_frequent')
city_imputed = imputer_mode.fit_transform(df[['city']])

异常值检测与处理

异常值可能是录入错误，也可能是真实但罕见的极端情况。处理方式取决于你对业务的理解。

IQR 方法

四分位距法是经典的非参数方法，不受极端值影响：

def detect_outliers_iqr(data, column):
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    outliers = data[(data[column] < lower_bound) | (data[column] > upper_bound)]
    return outliers, lower_bound, upper_bound

outliers, lb, ub = detect_outliers_iqr(df, 'age')
print(f'异常值数量: {len(outliers)}')
print(f'正常范围: [{lb:.1f}, {ub:.1f}]')

Z-score 方法

Z-score 假设数据近似正态分布：

from scipy import stats

def detect_outliers_zscore(data, column, threshold=3):
    z_scores = np.abs(stats.zscore(data[column].dropna()))
    outliers = data[column].dropna()[z_scores > threshold]
    return outliers

outliers_z = detect_outliers_zscore(df, 'income', threshold=3)
print(f'Z-score 异常值数量: {len(outliers_z)}')

可视化辅助

肉眼确认异常值是最可靠的方式：

fig, axes = plt.subplots(1, 3, figsize=(15, 4))

# 箱线图
axes[0].boxplot(df['age'].dropna())
axes[0].set_title('Age - Boxplot')

# 直方图
axes[1].hist(df['income'].dropna(), bins=50)
axes[1].set_title('Income - Histogram')

# 散点图
axes[2].scatter(df.index, df['score'])
axes[2].axhline(y=df['score'].mean() + 3*df['score'].std(), color='r', linestyle='--')
axes[2].axhline(y=df['score'].mean() - 3*df['score'].std(), color='r', linestyle='--')
axes[2].set_title('Score - Scatter with 3σ')

plt.tight_layout()
plt.show()

异常值处理策略

# 策略 1: 删除（确认是录入错误）
df_clean = df[~df.index.isin(outliers.index)]

# 策略 2: 截尾（限制到边界值）
df['age_capped'] = df['age'].clip(lower=lb, upper=ub)

# 策略 3: 视为缺失值，再用填充策略
df.loc[outliers.index, 'age'] = np.nan
df['age'] = df['age'].fillna(df['age'].median())

重复数据处理

重复数据会扭曲统计结果，必须处理：

# 检测完全重复的行
duplicates = df.duplicated()
print(f'完全重复行数: {duplicates.sum()}')

# 检测指定列重复
duplicates_subset = df.duplicated(subset=['age', 'income', 'city'])
print(f'基于指定列的重复行数: {duplicates_subset.sum()}')

# 查看重复行
df[df.duplicated(keep=False)].head()

# 删除重复行（保留第一个）
df_unique = df.drop_duplicates()

# 删除指定列重复（保留最后一条）
df_unique = df.drop_duplicates(subset=['id'], keep='last')

数据类型转换

数据类型不对会导致很多函数报错或结果异常：

# 查看各列数据类型
print(df.dtypes)

# 强制转换类型
df['id'] = df['id'].astype(str)         # int → str
df['age'] = df['age'].astype('Int64')   # 可空整数类型

# 数字字符串转数值
df['income_str'] = df['income'].apply(lambda x: f'¥{x:,.0f}')
df['income_clean'] = pd.to_numeric(
    df['income_str'].str.replace('¥', '').str.replace(',', ''),
    errors='coerce'
)

# 日期时间转换
df['signup_date'] = pd.to_datetime(df['signup_date'])
df['year'] = df['signup_date'].dt.year
df['month'] = df['signup_date'].dt.month
df['weekday'] = df['signup_date'].dt.day_name()

pd.to_numeric 的 errors='coerce' 参数非常实用——无法转换的值会被设为 NaN，不会中断流程。

字符串清洗

文本数据往往包含多余空格、不一致的大小写、特殊字符等：

# 去除首尾空格
df['city'] = df['city'].str.strip()

# 替换
df['phone_clean'] = df['phone'].str.replace('-', '')  # 去掉连字符
df['phone_clean'] = df['phone_clean'].str.replace(' ', '')

# 统一大小写
df['gender_std'] = df['gender'].str.upper()

# 正则表达式提取
df['phone_clean'] = df['phone'].str.extract(r'(\d{11})')

# 检测非标准格式
mask = df['phone'].str.match(r'^\d{11}$') == False
print(f'格式异常的号码: {mask.sum()}')

# 条件替换
df['city_std'] = df['city'].replace({
    '北京市': '北京',
    '上海': '上海',
    '上海市': '上海'
})

编码分类变量

大多数机器学习模型只能处理数值，分类变量需要编码。

Label Encoding（标签编码）

适合有序分类（如教育程度：小学 < 中学 < 大学）：

# Pandas 方式
df['gender_code'] = df['gender'].astype('category').cat.codes

# scikit-learn 方式
le = LabelEncoder()
df['city_label'] = le.fit_transform(df['city'].fillna('未知'))
print(dict(zip(le.classes_, le.transform(le.classes_))))

One-Hot Encoding（独热编码）

适合无序分类（如城市、颜色），避免模型错误地认为类别之间有大小关系：

# Pandas 方式
city_dummies = pd.get_dummies(df['city'], prefix='city', dummy_na=True)
df_encoded = pd.concat([df, city_dummies], axis=1)

# scikit-learn 方式
ohe = OneHotEncoder(sparse_output=False, handle_unknown='ignore')
city_ohe = ohe.fit_transform(df[['city']])
print(f'编码后特征数: {city_ohe.shape[1]}')

# 配合 ColumnTransformer 使用（生产环境推荐）
from sklearn.compose import ColumnTransformer

preprocessor = ColumnTransformer(transformers=[
    ('num', StandardScaler(), ['age', 'income', 'score']),
    ('cat', OneHotEncoder(), ['gender', 'city'])
])

何时用哪种？

编码方式	适用场景	缺点
Label Encoding	有序分类，树模型	给无序类别强加了顺序
One-Hot Encoding	无序分类，类别数少	类别一多维度爆炸
Target Encoding	高基数类别特征	容易过拟合，需要正则化

完整的清洗 Pipeline

把以上步骤整合成一个可复用的流水线：

from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer

# 数值特征 pipeline
numeric_features = ['age', 'income', 'score']
numeric_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# 分类特征 pipeline
categorical_features = ['gender', 'city']
categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# 组合
preprocessor = ColumnTransformer(transformers=[
    ('num', numeric_transformer, numeric_features),
    ('cat', categorical_transformer, categorical_features),
])

# 应用到数据
X_processed = preprocessor.fit_transform(df)
print(f'处理后数据形状: {X_processed.shape}')

实战：清洗一个真实的 CSV 文件

# 读取原始数据
df_raw = pd.read_csv('raw_data.csv')

# 第一步：概览
print('Shape:', df_raw.shape)
print('Info:')
df_raw.info()
print('Describe:')
print(df_raw.describe(include='all'))

# 第二步：处理缺失
missing_ratio = df_raw.isnull().mean()
cols_to_drop = missing_ratio[missing_ratio > 0.5].index
df_raw.drop(columns=cols_to_drop, inplace=True)
print(f'删除缺失超 50% 的列: {list(cols_to_drop)}')

# 第三步：处理异常值
for col in df_raw.select_dtypes(include=[np.number]).columns:
    Q1 = df_raw[col].quantile(0.25)
    Q3 = df_raw[col].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - 3 * IQR  # 放宽到 3 倍 IQR
    upper = Q3 + 3 * IQR
    df_raw[col] = df_raw[col].clip(lower, upper)

# 第四步：去重
df_raw.drop_duplicates(inplace=True)

# 第五步：类型修正
for col in df_raw.select_dtypes(include=['object']).columns:
    # 尝试转为数值
    df_raw[col] = pd.to_numeric(df_raw[col], errors='ignore')
    # 尝试转为日期
    try:
        df_raw[col] = pd.to_datetime(df_raw[col])
    except:
        pass

# 第六步：输出清洗报告
report = {
    '原始行数': len(df_raw) + df_raw.duplicated().sum() + 100,  # 模拟
    '最终行数': len(df_raw),
    '剩余列数': len(df_raw.columns),
    '缺失值比例': df_raw.isnull().mean().mean(),
}
print('清洗报告:', report)

# 保存清洗结果
df_raw.to_csv('clean_data.csv', index=False)

小结

数据清洗没有一成不变的公式，它依赖你对业务和数据的理解。本文覆盖了最核心的技术模块：

缺失值：检测 -> 分析缺失模式 -> 删除 / 填充 / 插值
异常值：IQR、Z-score 检测 -> 删除 / 截尾 / 标记
重复值：检测 -> 保留策略 -> 删除
类型转换：数值、日期、字符串的规范化
分类编码：Label vs One-Hot 的选择

下一篇文章中，我们会拿着清洗干净的数据，进入探索性数据分析（EDA）阶段。

Summary: 数据清洗六大核心模块与 Pandas 实战代码。