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[注册日期], errorscoerce) # 处理收入列的格式问题 df[收入] df[收入].replace(N/A, None) df[收入] df[收入].str.replace(,, ).astype(float) print(df.head())提示Pandas 2.2的errorscoerce参数在处理日期转换时非常有用它会将无法解析的值设为NaTNot 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[收入], errorscoerce) # 使用新的缺失值插补方法 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[注册月份]}, axis1 ) 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: fQ{x}) # 创建年龄分组 bins [0, 30, 40, 50, 100] labels [30, 30-39, 40-49, 50] viz_df[年龄组] pd.cut(viz_df[年龄], binsbins, labelslabels) # 创建收入分组 income_bins [0, 10000, 20000, 30000, float(inf)] income_labels [10k, 10-20k, 20-30k, 30k] viz_df[收入组] pd.cut(viz_df[收入], binsincome_bins, labelsincome_labels) # 保存清洗后的数据 viz_df.to_csv(cleaned_data.csv, indexFalse, encodingutf-8) print(viz_df.head())3.3 数据质量指标对比为了量化清洗效果我们可以对比清洗前后的关键指标质量指标清洗前清洗后改善幅度缺失值比例(%)16.70100%重复记录数10100%数据类型一致性50%100%100%异常值数量10100%日期格式标准化50%100%100%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, yNone): 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[注册日期], errorscoerce) # 处理收入列 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, downcastinteger) # 浮点数列使用最小兼容类型 float_cols df.select_dtypes(include[float64]).columns df[float_cols] df[float_cols].apply(pd.to_numeric, downcastfloat) # 对象类型转换为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())