Pandas 2.2 数据清洗3步法：从‘脏’CSV到可视化-ready的DataFrame

Pandas 2.2 数据清洗实战：从原始CSV到可视化就绪的完美转型

数据科学家的第一道关卡：理解脏数据本质

当我们从各种渠道获取原始数据时，经常会遇到令人头疼的数据质量问题。这些"脏数据"就像未经雕琢的玉石，需要经过精心打磨才能展现其真正价值。在数据分析项目中，数据清洗往往占据了整个流程70%以上的时间，这也是为什么专业的数据科学家必须掌握高效的数据清洗技术。

脏数据通常表现为以下几种形式：

• 缺失值：数据中的空白或占位符（如NaN、NULL）
• 异常值：超出合理范围的数值（如年龄为-1或200岁）
• 不一致格式：同一字段的不同表达方式（如日期格式混用）
• 重复记录：完全相同或高度相似的数据行
• 错误编码：字符编码问题导致的乱码
• 非标准分隔符：CSV文件中使用非逗号分隔符

# 典型脏数据示例 import pandas as pd data = { '姓名': ['张三', '李四', None, '王五', '赵六', '赵六'], '年龄': [25, -1, 30, 42, 35, 35], '收入': ['10,000', '15000', '20000', 'N/A', '25000', '25000'], '注册日期': ['2023-01-15', '2023/02/20', '2023-03-10', '15-04-2023', '2023-05-05', '2023-05-05'] } df = pd.DataFrame(data) print(df.head())

1. 数据质量评估与初步处理

1.1 数据质量诊断

在开始清洗前，我们需要全面了解数据的质量状况。Pandas 2.2提供了更强大的数据诊断工具：

# 数据质量快速诊断 def data_quality_report(df): # 缺失值统计 missing_values = df.isnull().sum() missing_percent = missing_values / len(df) * 100 # 数据类型统计 dtypes = df.dtypes # 唯一值统计 unique_values = df.nunique() # 创建质量报告DataFrame quality_report = pd.DataFrame({ '缺失值数量': missing_values, '缺失值比例(%)': missing_percent, '数据类型': dtypes, '唯一值数量': unique_values }) return quality_report.round(2) # 应用诊断函数 quality_report = data_quality_report(df) print(quality_report)

1.2 基础清洗操作

针对初步诊断结果，我们可以进行以下基础清洗：

# 处理重复数据 df = df.drop_duplicates() # 处理明显异常值 df.loc[df['年龄'] < 0, '年龄'] = None # 统一日期格式 df['注册日期'] = pd.to_datetime(df['注册日期'], errors='coerce') # 处理收入列的格式问题 df['收入'] = df['收入'].replace('N/A', None) df['收入'] = df['收入'].str.replace(',', '').astype(float) print(df.head())

提示：Pandas 2.2的errors='coerce'参数在处理日期转换时非常有用，它会将无法解析的值设为NaT（Not a Time），而不是抛出错误。

2. 高级数据转换技巧

2.1 使用Pandas 2.2新特性进行高效清洗

Pandas 2.2引入了几项提升数据清洗效率的新功能：

# 使用copy-on-write机制提升性能 pd.options.mode.copy_on_write = True # 减少不必要的数据复制 # 使用新的字符串方法链式操作 df['姓名'] = df['姓名'].str.strip().str.upper() # 使用新的数值转换方法 df['收入'] = pd.to_numeric(df['收入'], errors='coerce') # 使用新的缺失值插补方法 df['年龄'] = df['年龄'].fillna(df['年龄'].median()) df['收入'] = df['收入'].fillna(df.groupby('年龄')['收入'].transform('median')) print(df.head())

2.2 复杂数据转换实战

对于更复杂的数据转换需求，我们可以结合apply和lambda函数：

# 创建收入等级分类 df['收入等级'] = df['收入'].apply( lambda x: '低' if x < 15000 else ('中' if x < 25000 else '高') ) # 使用正则表达式提取和转换数据 import re df['注册年份'] = df['注册日期'].dt.year.astype(str) df['注册月份'] = df['注册日期'].dt.month.apply( lambda x: f'{x:02d}' ) # 创建复合ID df['用户ID'] = df.apply( lambda row: f"{row['姓名']}_{row['注册年份']}{row['注册月份']}", axis=1 ) print(df.head())

3. 数据质量验证与可视化准备

3.1 清洗后数据质量验证

完成清洗后，我们需要验证数据质量是否达到分析要求：

# 定义数据质量验证函数 def validate_data(df): # 检查缺失值 missing_check = df.isnull().sum().sum() == 0 # 检查数据类型 type_check = all([ pd.api.types.is_numeric_dtype(df['年龄']), pd.api.types.is_numeric_dtype(df['收入']), pd.api.types.is_datetime64_any_dtype(df['注册日期']) ]) # 检查唯一性 unique_check = df['用户ID'].is_unique # 综合验证结果 validation_result = { '无缺失值': missing_check, '数据类型正确': type_check, '用户ID唯一': unique_check, '通过所有检查': all([missing_check, type_check, unique_check]) } return validation_result # 执行验证 validation = validate_data(df) print(validation)

3.2 为可视化准备数据

清洗后的数据需要适当重构，以便更好地适应可视化需求：

# 创建可视化友好的数据格式 viz_df = df.copy() # 添加时间维度特征 viz_df['注册季度'] = viz_df['注册日期'].dt.quarter viz_df['注册季度'] = viz_df['注册季度'].apply(lambda x: f'Q{x}') # 创建年龄分组 bins = [0, 30, 40, 50, 100] labels = ['<30', '30-39', '40-49', '50+'] viz_df['年龄组'] = pd.cut(viz_df['年龄'], bins=bins, labels=labels) # 创建收入分组 income_bins = [0, 10000, 20000, 30000, float('inf')] income_labels = ['<10k', '10-20k', '20-30k', '30k+'] viz_df['收入组'] = pd.cut(viz_df['收入'], bins=income_bins, labels=income_labels) # 保存清洗后的数据 viz_df.to_csv('cleaned_data.csv', index=False, encoding='utf-8') print(viz_df.head())

3.3 数据质量指标对比

为了量化清洗效果，我们可以对比清洗前后的关键指标：

4. 自动化清洗流程设计

4.1 构建可复用的清洗管道

将上述步骤封装为可复用的函数和管道：

from sklearn.base import BaseEstimator, TransformerMixin class DataCleaner(BaseEstimator, TransformerMixin): def __init__(self): self.age_median = None self.income_median = None def fit(self, X, y=None): self.age_median = X['年龄'].median() self.income_median = X.groupby('年龄')['收入'].median() return self def transform(self, X): # 复制数据避免修改原始数据 X_clean = X.copy() # 处理重复数据 X_clean = X_clean.drop_duplicates() # 处理异常值 X_clean.loc[X_clean['年龄'] < 0, '年龄'] = None # 统一日期格式 X_clean['注册日期'] = pd.to_datetime(X_clean['注册日期'], errors='coerce') # 处理收入列 X_clean['收入'] = X_clean['收入'].replace('N/A', None) X_clean['收入'] = X_clean['收入'].str.replace(',', '').astype(float) # 填充缺失值 X_clean['年龄'] = X_clean['年龄'].fillna(self.age_median) X_clean['收入'] = X_clean['收入'].fillna( X_clean['年龄'].map(self.income_median) ) return X_clean # 使用清洗管道 cleaner = DataCleaner() cleaner.fit(df) cleaned_df = cleaner.transform(df)

4.2 集成测试与异常处理

为确保清洗流程的健壮性，需要添加全面的测试和异常处理：

def test_data_cleaner(): # 创建测试数据 test_data = { '姓名': ['测试1', '测试2', None], '年龄': [20, -5, 30], '收入': ['5,000', 'N/A', '20000'], '注册日期': ['2023-01-01', 'invalid', '2023-03-01'] } test_df = pd.DataFrame(test_data) # 应用清洗器 cleaner = DataCleaner() cleaner.fit(test_df) cleaned = cleaner.transform(test_df) # 断言测试 assert cleaned.isnull().sum().sum() == 0, "存在未处理的缺失值" assert all(cleaned['年龄'] >= 0), "存在负年龄" assert pd.api.types.is_numeric_dtype(cleaned['收入']), "收入不是数值类型" assert pd.api.types.is_datetime64_any_dtype(cleaned['注册日期']), "日期类型不正确" print("所有测试通过!") # 运行测试 test_data_cleaner()

4.3 性能优化技巧

处理大型数据集时，性能优化至关重要：

# 使用高效数据类型 def optimize_dtypes(df): # 整数列使用最小兼容类型 int_cols = df.select_dtypes(include=['int64']).columns df[int_cols] = df[int_cols].apply(pd.to_numeric, downcast='integer') # 浮点数列使用最小兼容类型 float_cols = df.select_dtypes(include=['float64']).columns df[float_cols] = df[float_cols].apply(pd.to_numeric, downcast='float') # 对象类型转换为category（如果基数低） for col in df.select_dtypes(include=['object']).columns: if df[col].nunique() / len(df) < 0.5: # 唯一值比例小于50% df[col] = df[col].astype('category') return df # 应用优化 optimized_df = optimize_dtypes(cleaned_df) print(optimized_df.info())