📚 课程目录第3章实现数据质量检查流水线

3. 实现数据质量检查流水线

任务目标

构建一个可复用的数据质量检查流水线，自动检测数据集中的缺失值、重复记录和异常值（基于Z-score），并输出结构化的质量报告（JSON格式），为后续训练流程提供数据准入依据。

环境与依赖

Python 3.8+，仅使用标准库（csv, json, statistics, datetime），无需第三方包安装。

核心实现

import csv
import json
import statistics
from datetime import datetime
from pathlib import Path

class DataQualityPipeline:
    def __init__(self, numeric_threshold=3.0):
        self.numeric_threshold = numeric_threshold  # Z-score异常阈值

    def _load_csv(self, filepath):
        """读取CSV，返回(表头, 行数据列表)"""
        with open(filepath, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            rows = list(reader)
        return reader.fieldnames, rows

    def _check_missing(self, rows, fields):
        """检测缺失值，返回{字段: 缺失数量}"""
        missing = {}
        for field in fields:
            count = sum(1 for r in rows if r.get(field, '').strip() == '')
            if count > 0:
                missing[field] = count
        return missing

    def _check_duplicates(self, rows):
        """检测完全重复的行，返回重复行数"""
        seen = set()
        dup_count = 0
        for r in rows:
            key = tuple(r.items())
            if key in seen:
                dup_count += 1
            else:
                seen.add(key)
        return dup_count

    def _check_outliers(self, rows, fields):
        """对数值字段用Z-score检测异常值，返回{字段: [异常行索引]}"""
        outliers = {}
        for field in fields:
            values = []
            valid_rows = []
            for idx, r in enumerate(rows):
                try:
                    val = float(r[field])
                    values.append(val)
                    valid_rows.append(idx)
                except (ValueError, KeyError):
                    continue
            if len(values) < 3:  # 样本太少不检测
                continue
            mean = statistics.mean(values)
            stdev = statistics.stdev(values)
            if stdev == 0:
                continue
            bad_idx = [valid_rows[i] for i, v in enumerate(values)
                       if abs(v - mean) / stdev > self.numeric_threshold]
            if bad_idx:
                outliers[field] = bad_idx
        return outliers

    def run(self, csv_path):
        """执行完整检查，返回报告dict"""
        fields, rows = self._load_csv(csv_path)
        report = {
            "file": str(csv_path),
            "timestamp": datetime.now().isoformat(timespec='seconds'),
            "total_rows": len(rows),
            "total_fields": len(fields),
            "missing": self._check_missing(rows, fields),
            "duplicate_rows": self._check_duplicates(rows),
            "outliers": self._check_outliers(rows, fields),
            "passed": True
        }
        # 判定标准：无缺失、无重复、无异常
        if report["missing"] or report["duplicate_rows"] > 0 or report["outliers"]:
            report["passed"] = False
        return report

    def run_and_save(self, csv_path, output_path):
        """运行并保存JSON报告"""
        report = self.run(csv_path)
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        return report

# 最小调用入口
if __name__ == "__main__":
    import sys
    if len(sys.argv) != 3:
        print("用法: python quality_pipeline.py <输入.csv> <输出.json>")
        sys.exit(1)
    pipeline = DataQualityPipeline()
    result = pipeline.run_and_save(sys.argv[1], sys.argv[2])
    print(f"检查完成: {'通过' if result['passed'] else '发现问题'}")
    print(f"报告已保存至: {sys.argv[2]}")

运行方式

将上述代码保存为 quality_pipeline.py。
准备待检查的CSV文件（需包含表头）。
执行命令：python quality_pipeline.py data.csv report.json

验证方法

使用以下样例数据创建 sample.csv 并运行，观察输出报告：

id,name,age,salary
1,Alice,25,5000
2,Bob,30,6000
3,Charlie,,7000
4,David,35,8000
5,Eve,40,9000
1,Alice,25,5000
6,Frank,200,100000

预期报告内容：missing 包含 age 字段缺失1条；duplicate_rows 为1（第5行重复第0行）；outliers 中 age 字段包含行索引5（值200），salary 字段包含行索引5（值100000）；passed 为 false。打开生成的 report.json 核对上述字段是否与预期一致。

常见问题

非数值字段：异常值检测仅对可转换为float的字段生效，无法转换的字段自动跳过。
小样本：少于3个有效数值的字段不进行异常检测，避免统计意义不足。
编码问题：CSV需为UTF-8编码，否则读取可能报错。