数据清洗题

题目背景

理解面试官出这道题的意图

这道题考察的是数据处理能力和工程规范意识。在测试开发工作中，经常需要处理原始数据：从生产导出的数据、从日志提取的数据、从第三方系统获取的数据。这些数据往往存在缺失、异常、格式不一致等问题，直接用于测试会导致结果不可靠。面试官想了解你是否能设计出一个结构清晰、可追溯、可复用的数据清洗流程，而不是写一段一次性脚本。工程规范意识体现在对原始数据的保护、清洗过程的记录、异常数据的分类处理和结果的校验验证。

解题思路

规则定义→清洗执行→异常处理→结果校验

• 第一步：规则定义——在清洗前明确定义清洗规则，而不是边写边改。缺失值处理策略：填充默认值、使用均值/中位数、标记为特殊值、直接丢弃。异常值处理策略：修正为合理范围、标记为异常、单独分析。重复值处理策略：保留第一条、保留最新、合并去重。格式不一致处理策略：统一日期格式、统一编码、统一单位。规则定义应文档化，方便团队理解和复用。
• 第二步：清洗执行——每条清洗操作独立执行，有明确的输入和输出。清洗步骤按顺序执行，前一步的输出作为后一步的输入。每步操作记录日志：操作类型、影响行数、处理前后的样本数据。清洗逻辑拆分为独立函数，每个函数只负责一种清洗操作（如去除重复、填充缺失、格式统一）。
• 第三步：异常处理——异常数据分类处理，不简单地全部丢弃。可修正的异常：格式错误但可推断正确值（如日期格式不统一），自动修正。需人工确认的异常：值超出合理范围但可能是真实数据（如金额异常大），标记后人工复核。无法处理的异常：数据严重损坏或缺失关键字段，记录到异常报告后丢弃。每种异常类型统计数量，生成异常报告。
• 第四步：结果校验——清洗完成后验证输出是否符合预期。格式校验：检查输出数据的格式是否符合要求（字段类型、日期格式、编码）。约束校验：检查业务约束是否满足（如金额非负、日期合理、外键关联存在）。抽样检查：随机抽取部分数据人工核对，确认清洗结果正确。校验失败时定位问题步骤，回退或修正。
• 设计权衡：清洗严格度 vs 数据保留率、自动化程度 vs 人工介入、处理速度 vs 准确性，需要根据数据用途做选择。

代码逻辑

核心流程描述，不展示完整代码

【整体流程】读取原始数据 → 保存原始数据快照 → 按顺序执行清洗步骤 → 每步记录清洗日志 → 收集异常数据 → 执行结果校验 → 输出清洗后数据和异常报告。【核心步骤详解】1. 数据读取器：从文件（CSV、Excel、JSON）或数据库读取原始数据。统一转换为内部数据结构（如 DataFrame 或对象列表）。读取时记录数据量、字段列表、样本数据。2. 原始数据快照：清洗前保存原始数据的副本或哈希值。用于对比清洗前后的变化，以及问题回溯时恢复原始数据。快照存储路径与清洗日志关联。3. 清洗步骤执行器：按预定义的清洗步骤列表依次执行。每个步骤是一个独立函数，接收数据返回清洗后的数据。步骤之间通过数据管道传递，前一步输出作为后一步输入。4. 异常数据收集器：清洗过程中发现的异常数据不直接丢弃，而是分类收集。按异常类型分组（格式错误、缺失关键字段、超出范围），记录异常原因和原始值。5. 结果校验器：清洗完成后执行校验。格式校验：检查每个字段的类型和格式。约束校验：检查业务规则是否满足。数据量校验：检查清洗后数据量是否在合理范围（不应大幅减少）。6. 报告生成器：生成清洗报告，包含：原始数据量、清洗后数据量、丢弃数据量、异常分类统计、每步清洗的影响行数、校验结果。【关键接口定义】CleaningRule：定义清洗操作类型、参数、适用字段。CleaningStep：封装单个清洗步骤的执行逻辑和日志记录。CleaningReport：记录清洗全过程的统计信息和异常数据。

示例代码：数据清洗

# utils/data_cleaner.py - 数据清洗工具
import logging
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

@dataclass
class CleaningStats:
    total: int = 0
    cleaned: int = 0
    dropped: int = 0
    issues: list[str] = field(default_factory=list)

class DataCleaner:
    def __init__(self):
        self.stats = CleaningStats()

    def clean(self, data: list[dict], rules: dict) -> list[dict]:
        """执行数据清洗"""
        self.stats.total = len(data)
        cleaned = []

        for row in data:
            try:
                result = self._apply_rules(row, rules)
                if result is not None:
                    cleaned.append(result)
                    self.stats.cleaned += 1
            except ValueError as e:
                self.stats.dropped += 1
                self.stats.issues.append(str(e))
                logger.warning(f"丢弃数据: {e}")

        logger.info(
            f"清洗完成: 总数={self.stats.total}, "
            f"有效={self.stats.cleaned}, 丢弃={self.stats.dropped}"
        )
        return cleaned

    def _apply_rules(self, row: dict, rules: dict) -> dict | None:
        """应用清洗规则"""
        # 必填字段检查
        for field_name in rules.get("required", []):
            if not row.get(field_name):
                raise ValueError(f"缺少必填字段: {field_name}")

        # 字符串修剪
        cleaned = {
            k: v.strip() if isinstance(v, str) else v
            for k, v in row.items()
        }

        # 数值范围校验
        for fname, (min_v, max_v) in rules.get("ranges", {}).items():
            val = cleaned.get(fname)
            if val is not None and not (min_v <= val <= max_v):
                raise ValueError(f"{fname}={val} 超出范围")

        return cleaned

# 使用示例
cleaner = DataCleaner()
rules = {
    "required": ["name", "email"],
    "ranges": {"age": (0, 150)},
}
raw_data = [
    {"name": "  Alice  ", "email": "alice@test.com", "age": 25},
    {"name": "", "email": "bob@test.com"},  # 缺少 name
    {"name": "Charlie", "email": "c@test.com", "age": 200},  # 超限
]
result = cleaner.clean(raw_data, rules)
print(f"有效数据: {len(result)} 条, 统计: {cleaner.stats}")

常见失分点

面试中最容易丢分的 5 个问题

失分点 1：清洗逻辑写在大段代码里

错误做法：把所有清洗逻辑写在一个函数或脚本中，没有拆分和注释。

为什么不好：代码难以理解和维护。新增清洗规则时需要修改大段代码，容易引入 Bug。团队协作时其他人无法快速理解清洗流程。

如何改进：每个清洗操作独立为函数，函数名清晰描述操作内容（如 remove_duplicates、fill_missing_values、normalize_date_format）。主流程按顺序调用各清洗函数，逻辑清晰。

失分点 2：没有保留原始数据和清洗日志

错误做法：直接修改原始数据，不保留副本，不记录清洗过程。

为什么不好：清洗出错时无法恢复原始数据。无法追溯某条数据是如何被处理的。排查问题时没有依据。

如何改进：清洗前保存原始数据快照。每步清洗操作记录日志：操作类型、影响行数、处理前后的样本数据。日志与清洗结果关联存储。

失分点 3：异常数据直接丢弃不做统计

错误做法：发现异常数据直接删除，不记录异常类型和数量。

为什么不好：无法了解数据质量。异常数据可能包含重要信息（如业务异常信号）。丢弃比例过高时无法察觉。

如何改进：异常数据分类收集，记录异常类型、原始值、丢弃原因。生成异常报告，包含各类异常的数量和占比。异常比例过高时发出警告。

失分点 4：清洗后没有校验步骤

错误做法：清洗完成后直接使用结果，不验证数据质量。

为什么不好：清洗逻辑可能有 Bug，导致结果数据不符合预期。格式错误、数据丢失等问题在使用时才会发现。

如何改进：清洗完成后执行校验：格式校验、约束校验、数据量校验。抽样检查部分数据的人工核对。校验失败时定位问题步骤并修正。

失分点 5：清洗规则不可配置

错误做法：清洗规则硬编码在代码中，无法根据不同数据源调整。

为什么不好：不同数据源的清洗需求不同，硬编码规则无法复用。规则变更需要修改代码，重新部署。

如何改进：清洗规则配置化。使用配置文件或参数定义清洗规则（如缺失值填充策略、异常值判定阈值）。清洗引擎读取配置后执行，规则变更只需修改配置。

进阶讨论

展示技术深度和系统思维

【数据清洗管道模式】使用管道模式组织清洗步骤，每个步骤是独立的管道节点。优势：步骤可复用、可组合、可测试。新增步骤只需添加到管道中，不影响现有步骤。支持条件执行，根据数据特征动态选择清洗步骤。【大数据量处理】数据量大时（百万级以上），需要考虑性能。分批处理：将数据分块，每块独立清洗后合并。流式处理：使用生成器逐行处理，避免全量加载到内存。并行处理：无依赖的清洗步骤并行执行。【数据质量度量】建立数据质量度量体系，量化清洗效果。指标：完整率（非空字段比例）、准确率（符合业务规则的比例）、一致率（格式统一的比例）、及时率（数据更新延迟）。清洗前后对比这些指标，评估清洗效果。