数据清洗通常被视为数据科学家和分析师在进行“真正”的分析之前必须经历的手动、耗时的过程。
然而，借助像pandas这样的Python库，我们可以自动化许多常见的清洗任务，从而创建一个可靠、可重复的管道。关键在于识别常见的数据质量问题，并实施能够一致处理各种数据集的系统解决方案。
在本文中，我们将探讨构建这样一个自动化数据清洗管道的步骤。
▶️ 链接到Google Colab笔记本
首先，导入pandas：
import pandas as pd步骤1：运行基本数据质量检查
在开始任何清洗之前，我们需要了解所处理数据的质量。因此，第一步是评估数据的当前状态。
我们需要识别：
每列中的缺失值重复行基本数据特征让我们从一些基本质量检查开始：
def check_data_quality(df): # 存储初始数据质量指标 quality_report = { 'missing_values': df.isnull().sum().to_dict(), 'duplicates': df.duplicated().sum(), 'total_rows': len(df), 'memory_usage': df.memory_usage().sum() / 1024**2 # 以MB为单位 } return quality_report这为我们提供了数据质量的基本了解，并帮助我们识别需要执行的特定清洗任务。
步骤2：标准化数据类型
原始数据中最常见的问题之一是数据类型不一致。例如，日期可能存储为字符串，或者数值可能包含货币符号等。
因此，下一步是确保所有字段都具备正确的/预期的数据类型。这包括：
将字符串日期转换为datetime对象识别并将数值字符串转换为实际数字确保分类变量正确编码def standardize_datatypes(df): for column in df.columns: # 尝试将字符串日期转换为datetime if df[column].dtype == 'object': try: df[column] = pd.to_datetime(df[column]) print(f"将{column}转换为datetime") except ValueError: # 如果datetime转换失败，尝试转换为数值 try: df[column] = pd.to_numeric(df[column].str.replace('$', '').str.replace(',', '')) print(f"将{column}转换为数值") except: pass return df此步骤可以防止后续分析中出现类型相关的错误。
步骤3：处理缺失值
缺失值会显著影响我们的分析。与其删除包含缺失值的数据记录，我们可以使用插补策略：
对数值列使用中位数插补对分类数据应用众数插补在填补空白的同时保持数据集的统计属性以下是我们如何在数值和分类列中插补缺失值的方法：
from sklearn.impute import SimpleImputerdef handle_missing_values(df): # 处理数值列 numeric_columns = df.select_dtypes(include=['int64', 'float64']).columns if len(numeric_columns) > 0: num_imputer = SimpleImputer(strategy='median') df[numeric_columns] = num_imputer.fit_transform(df[numeric_columns]) # 处理分类列 categorical_columns = df.select_dtypes(include=['object']).columns if len(categorical_columns) > 0: cat_imputer = SimpleImputer(strategy='most_frequent') df[categorical_columns] = cat_imputer.fit_transform(df[categorical_columns]) return df在这里，我们使用了Scikit-learn的SimpleImputer类，但你也可以使用pandas中的插补函数。
步骤4：检测并处理异常值
异常值可能会使我们的分析产生偏差，因此我们需要谨慎处理。
⚠️ 你需要领域知识来决定“什么”可能实际上是异常值。
以下是使用四分位距（IQR）方法的一种方法：
计算数值列的四分位距（IQR）识别超出四分位数1.5 * IQR的值对极值进行封顶处理，而不是删除它们这可以在管理极值的同时保留数据。
def remove_outliers(df): numeric_columns = df.select_dtypes(include=['int64', 'float64']).columns outliers_removed = {} for column in numeric_columns: Q1 = df[column].quantile(0.25) Q3 = df[column].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR # 在移除之前计算异常值数量 outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)].shape[0] # 对值进行封顶处理，而不是删除它们 df[column] = df[column].clip(lower=lower_bound, upper=upper_bound) if outliers > 0: outliers_removed[column] = outliers return df, outliers_removed你可以了解更多处理异常值的常见技术。
步骤5：验证结果
清洗后，我们需要验证管道是否按预期工作：
确认没有剩余的缺失值检查是否有剩余的重复项验证数据完整性和一致性生成全面的清洗报告def validate_cleaning(df, original_shape, cleaning_report): validation_results = { 'rows_remaining': len(df), 'missing_values_remaining': df.isnull().sum().sum(), 'duplicates_remaining': df.duplicated().sum(), 'data_loss_percentage': (1 - len(df)/original_shape[0]) * 100 } # 将验证结果添加到清洗报告中 cleaning_report['validation'] = validation_results return cleaning_report最后，让我们将所有内容整合到一个完整的管道中：
def automated_cleaning_pipeline(df): # 存储原始形状用于报告 original_shape = df.shape # 初始化清洗报告 cleaning_report = {} # 执行每个步骤并收集指标 cleaning_report['initial_quality'] = check_data_quality(df) df = standardize_datatypes(df) df = handle_missing_values(df) df, outliers = remove_outliers(df) cleaning_report['outliers_removed'] = outliers # 验证并最终化报告 cleaning_report = validate_cleaning(df, original_shape, cleaning_report) return df, cleaning_report现在，你可以在简单的样本数据上测试你的管道。
总结
正如你所看到的，自动化数据清洗不仅节省时间，还确保了数据准备过程中的一致性和可重复性。我分享的管道处理了常见的数据质量问题，同时提供了有关所做更改的详细报告。
你可能需要根据你的领域知识和特定要求调整清洗策略。
请记住，虽然自动化很有帮助，但你应该始终查看清洗报告，并根据你的具体用例验证结果。祝你数据清洗愉快！
