Python 构建数据清理和验证管道完整版

数

云朵君2025年11月03日约 10 分钟

数据质量是任何数据科学项目的基石。数据质量差会导致错误的模型、误导性的见解以及代价高昂的业务决策。在本指南中，我们将探索如何使用 Python 构建强大而简洁的数据清理和验证流程。

什么是数据清理和验证管道？

数据清理和验证流程是一种自动化的工作流程，它系统地处理原始数据，以确保其质量在进行分析之前符合可接受的标准。可以将其视为数据的质量控制系统：

检测和处理缺失值——检测数据集中的缺口并应用适当的处理策略
验证数据类型和格式——确保每个字段都包含预期类型的信息
识别并删除异常值——检测可能影响分析的异常值
执行业务规则——应用特定领域的约束和验证逻辑
维护血统——追踪进行了哪些转换以及何时进行

管道本质上充当守门人的角色，以确保只有干净且经过验证的数据才能流入你的分析和机器学习工作流程。

为什么要使用数据清理管道？

自动清洁管道的一些主要优点包括：

一致性和可重复性：手动方法可能会在清洁过程中引入人为错误和不一致性。自动化流水线一遍又一遍地执行相同的清洁逻辑，从而使结果具有可重复性和可信度。
时间和资源效率：准备数据可能占用数据科学家 70% 到 80% 的时间。管道可以自动化数据清理流程，大大减少这方面的开销，从而引导团队专注于分析和建模。
可扩展性：例如，随着数据量的增长，手动清理变得难以为继。管道可以优化大型数据集的处理，并几乎自动地应对不断增长的数据负载。
减少错误：自动验证可以发现手动检查可能遗漏的数据质量问题，从而降低从伪造数据得出错误结论的风险。
审计跟踪：现有的管道为你精确概述了清理数据所遵循的步骤，这在法规遵从和调试方面非常有用。

设置开发环境

在开始构建管道之前，请确保我们拥有所有工具。我们的管道将利用Python强大的库：

import pandas as pd
import numpy as np
from datetime import datetime
import logging
from typing import Dict, List, Any, Optional

为什么是这些图书馆？

代码中将使用以下库，然后是它们提供的实用程序：

pandas：稳健地操作和分析数据
numpy：提供快速的数值运算和数组处理
datetime：验证并格式化日期和时间
logging：启用管道执行和错误跟踪以进行调试
打字：实际上增加了代码文档的类型提示并避免了常见错误

定义验证架构

验证模式本质上是一份蓝图，它定义了数据所基于的结构及其遵循的约束的期望。我们的模式定义如下：

VALIDATION_SCHEMA = {
'user_id': {'type': int, 'required': True, 'min_value': 1},
'email': {'type': str, 'required': True, 'pattern': r'^[^@]+@[^@]+\.[^@]+$'},
'age': {'type': int, 'required': False, 'min_value': 0, 'max_value': 120},
'signup_date': {'type': 'datetime', 'required': True},
'score': {'type': float, 'required': False, 'min_value': 0.0, 'max_value': 100.0}

该模式指定了许多验证规则：

类型验证：检查每个字段接收值的数据类型
必填字段验证：标识不可缺少的必填字段
范围验证：设置可接受的最小值和最大值
模式验证：用于验证目的的正则表达式，例如有效的电子邮件地址
日期验证：检查日期字段是否包含有效的日期时间对象

构建管道类

我们的管道类将充当协调所有清理和验证操作的协调器：

classDataCleaningPipeline:
def__init__(self, schema: Dict[str, Any]):
        self.schema = schema
        self.errors = []
        self.cleaned_rows = 0
        self.total_rows = 0
# Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
defclean_and_validate(self, df: pd.DataFrame) -> pd.DataFrame:
"""Main pipeline orchestrator"""
        self.total_rows = len(df)
        self.logger.info(f"Starting pipeline with {self.total_rows} rows")
# Pipeline stages
        df = self._handle_missing_values(df)
        df = self._validate_data_types(df)
        df = self._apply_constraints(df)
        df = self._remove_outliers(df)
        self.cleaned_rows = len(df)
        self._generate_report()
return df

该管道遵循系统方法：

初始化跟踪变量来监控清洁进度
设置日志记录以捕获管道执行详细信息
按逻辑顺序执行清洁阶段
生成总结清洁结果的报告

编写数据清理逻辑

让我们通过强大的错误处理来实现每个清理阶段：

缺失值处理

以下代码将删除缺少必填字段的行，并使用中位数（对于数字）或“未知”（对于非数字）填充缺少的可选字段。

def_handle_missing_values(self, df: pd.DataFrame) -> pd.DataFrame:
"""Handle missing values based on field requirements"""
for column, rules in self.schema.items():
if column in df.columns:
if rules.get('required', False):
# Remove rows with missing required fields
                missing_count = df[column].isnull().sum()
if missing_count > 0:
                    self.errors.append(f"Removed {missing_count} rows with missing {column}")
                    df = df.dropna(subset=[column])
else:
# Fill optional missing values
if df[column].dtype in ['int64', 'float64']:
                    df[column].fillna(df[column].median(), inplace=True)
else:
                    df[column].fillna('Unknown', inplace=True)
return df

数据类型验证

以下代码将列转换为指定类型并删除转换失败的行。

def_validate_data_types(self, df: pd.DataFrame) -> pd.DataFrame:
"""Convert and validate data types"""
for column, rules in self.schema.items():
if column in df.columns:
            expected_type = rules['type']
try:
if expected_type == 'datetime':
                    df[column] = pd.to_datetime(df[column], errors='coerce')
elif expected_type == int:
                    df[column] = pd.to_numeric(df[column], errors='coerce').astype('Int64')
elif expected_type == float:
                    df[column] = pd.to_numeric(df[column], errors='coerce')
# Remove rows with conversion failures
                invalid_count = df[column].isnull().sum()
if invalid_count > 0:
                    self.errors.append(f"Removed {invalid_count} rows with invalid {column}")
                    df = df.dropna(subset=[column])
except Exception as e:
                self.logger.error(f"Type conversion error for {column}: {e}")
return df

添加带有错误跟踪的验证

我们的约束验证系统确保数据在限制范围内并且格式可接受：

def_apply_constraints(self, df: pd.DataFrame) -> pd.DataFrame:
"""Apply field-specific constraints"""
for column, rules in self.schema.items():
if column in df.columns:
            initial_count = len(df)
# Range validation
if'min_value'in rules:
                df = df[df[column] >= rules['min_value']]
if'max_value'in rules:
                df = df[df[column] <= rules['max_value']]
# Pattern validation for strings
if'pattern'in rules and df[column].dtype == 'object':
import re
                pattern = re.compile(rules['pattern'])
                df = df[df[column].astype(str).str.match(pattern, na=False)]
            removed_count = initial_count - len(df)
if removed_count > 0:
                self.errors.append(f"Removed {removed_count} rows failing {column} constraints")
return df

基于约束和跨字段验证

当考虑多个字段之间的关系时，通常需要高级验证：

def_cross_field_validation(self, df: pd.DataFrame) -> pd.DataFrame:
"""Validate relationships between fields"""
    initial_count = len(df)
# Example: Signup date should not be in the future
if'signup_date'in df.columns:
        future_signups = df['signup_date'] > datetime.now()
        df = df[~future_signups]
        removed = future_signups.sum()
if removed > 0:
            self.errors.append(f"Removed {removed} rows with future signup dates")
# Example: Age consistency with signup date
if'age'in df.columns and'signup_date'in df.columns:
# Remove records where age seems inconsistent with signup timing
        suspicious_age = (df['age'] < 13) & (df['signup_date'] < datetime(2010, 1, 1))
        df = df[~suspicious_age]
        removed = suspicious_age.sum()
if removed > 0:
            self.errors.append(f"Removed {removed} rows with suspicious age/date combinations")
return df

异常值检测和去除

异常值对分析结果的影响可能非常大。该流程提供了一种先进的方法来检测此类异常值：

def_remove_outliers(self, df: pd.DataFrame) -> pd.DataFrame:
"""Remove statistical outliers using IQR method"""
    numeric_columns = df.select_dtypes(include=[np.number]).columns
for column in numeric_columns:
if column in self.schema:
            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[column] < lower_bound) | (df[column] > upper_bound)
            outlier_count = outliers.sum()
if outlier_count > 0:
                df = df[~outliers]
                self.errors.append(f"Removed {outlier_count} outliers from {column}")
return df

编排管道

这是我们完整、紧凑的管道实现：

classDataCleaningPipeline:
def__init__(self, schema: Dict[str, Any]):
        self.schema = schema
        self.errors = []
        self.cleaned_rows = 0
        self.total_rows = 0
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

defclean_and_validate(self, df: pd.DataFrame) -> pd.DataFrame:
        self.total_rows = len(df)
        self.logger.info(f"Starting pipeline with {self.total_rows} rows")
# Execute cleaning stages
        df = self._handle_missing_values(df)
        df = self._validate_data_types(df)
        df = self._apply_constraints(df)
        df = self._remove_outliers(df)
        self.cleaned_rows = len(df)
        self._generate_report()
return df
def_generate_report(self):
"""Generate cleaning summary report"""
        self.logger.info(f"Pipeline completed: {self.cleaned_rows}/{self.total_rows} rows retained")
for error in self.errors:
            self.logger.warning(error)

示例用法

使用真实数据集的管道的演示：

# Create sample problematic data
sample_data = pd.DataFrame({
'user_id': [1, 2, None, 4, 5, 999999],
'email': ['[email protected]', 'invalid-email', '[email protected]', None, '[email protected]', '[email protected]'],
'age': [25, 150, 30, -5, 35, 28],  # Contains invalid ages
'signup_date': ['2023-01-15', '2030-12-31', '2022-06-10', '2023-03-20', 'invalid-date', '2023-05-15'],
'score': [85.5, 105.0, 92.3, 78.1, -10.0, 88.7]  # Contains out-of-range scores
})
# Initialize and run pipeline
pipeline = DataCleaningPipeline(VALIDATION_SCHEMA)
cleaned_data = pipeline.clean_and_validate(sample_data)
print("Cleaned Data:")
print(cleaned_data)
print(f"\nCleaning Summary: {pipeline.cleaned_rows}/{pipeline.total_rows} rows retained")

输出：

输出显示了最终清理后的 DataFrame，其中删除了缺少必填字段、数据类型无效、违反约束条件（例如超出范围的值或错误的电子邮件地址）以及包含异常值的行。摘要行报告了在总数中保留了多少行。这确保只有有效的、可供分析的数据才能继续处理，从而提高质量、减少错误，并使你的流程可靠且可重复。

扩展管道

我们的管道已实现可扩展。以下是一些改进建议：

自定义验证规则：通过扩展模式格式来接受自定义验证功能，从而合并特定于域的验证逻辑。
并行处理：使用适当的库（例如多处理）在多个 CPU 核心上并行处理大型数据集。
机器学习集成：引入异常检测模型来检测基于规则的系统无法解决的数据质量问题。
实时处理：使用Apache Kafka或Apache Spark Streaming修改流数据管道。
数据质量指标：设计一个广泛的质量分数，考虑完整性、准确性、一致性和及时性等多个维度。

结论

这种清理和验证的理念是检查数据中所有可能出错的元素：缺失值、无效的数据类型或约束、异常值，当然，还要尽可能详细地报告所有这些信息。之后，此流程将成为你在任何数据分析或机器学习任务中进行数据质量保证的起点。这种方法的优势包括：自动 QA 检查（确保不会遗漏任何错误）、可重现的结果、全面的错误跟踪，以及在特定领域约束下轻松安装多项检查。

通过在数据工作流中部署此类管道，你的数据驱动决策将更有可能保持正确性和精准性。数据清理是一个迭代过程，随着新的数据质量问题出现，你可以在你的领域中扩展此管道，添加额外的验证规则和清理逻辑。这种模块化设计允许集成新功能，而不会与当前已实现的功能发生冲突。

🏴‍☠️宝藏级🏴‍☠️ 原创公众号『数据STUDIO』内容超级硬核。公众号以Python为核心语言，垂直于数据科学领域，包括可戳👉Python｜MySQL｜数据分析｜数据可视化｜机器学习与数据挖掘｜爬虫等，从入门到进阶！

长按👇关注- 数据STUDIO -设为星标，干货速递

目录