当前位置：首页 > Python 工业工具 > 正文内容

数据清洗实战：来自MES的脏数据怎么办

一、问题背景：数据90%的时间花在清洗上

去年我做了一个数据分析项目，以为重点是分析。结果发现80%的时间都在做数据清洗。

典型的MES导出数据：

Lot ID,Thickness,Wafer Count,Process,Date FAB-001,1250.5,25,ETCH,2026/01/15 FAB-002,,"24",ETCH,2026-01-15 FAB-003,N/A,25,etch,2026.01.15 FAB-004,-999,25,"ETCH ",2026/01/15

看到问题了吗？

1. FAB-002：厚度为空

2. FAB-003：厚度写N/A，日期格式不同

3. FAB-004：厚度为负值（不合理），工序名有空格

4. 日期格式有三种

这些数据在Excel里看起来没问题，但导入Python就会报错或者算错。

---

二、技术原理：数据清洗的原则

2.1 清洗流程

def clean_pipeline(raw_data): """标准数据清洗流程""" data = raw_data.copy() # Step 1: 格式标准化 data = standardize_formats(data) # Step 2: 处理缺失值 data = handle_missing_values(data) # Step 3: 处理异常值 data = handle_outliers(data) # Step 4: 去重 data = remove_duplicates(data) # Step 5: 验证 data = validate_data(data) return data

2.2 常见的脏数据

脏数据类型 | 描述 | 例子

|-----------|------|------|

缺失值 | 该有的数据没有 | 空单元格、null、N/A

异常值 | 不合理的数据 | 负厚度、99999

格式不一致 | 同一字段不同格式 | 2026/01/01 vs 2026-01-01

重复数据 | 相同数据出现多次 | 同一Lot出现两次

拼写错误 | 字段内容不一致 | ETCH vs etch vs Etch_

---

三、实战案例：MES数据清洗器

""" MES数据清洗系统功能：自动清洗MES导出的脏数据 """ import pandas as pd import numpy as np import re from datetime import datetime from typing import List, Dict, Optional, Tuple import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class MESDataCleaner: """MES数据清洗器 - 核心清洗逻辑""" def __init__(self): self.cleaning_log = [] # 记录清洗日志，便于追溯问题 def clean_thickness(self, value) -> Optional[float]: """ 清洗厚度数据为什么这样写： - 厚度是关键参数，任何异常都会影响后续分析 - 支持多种无效值格式（N/A、空字符串、负值） - 自动移除单位（A、nm等），避免类型错误 """ original = value # 先处理None，避免后续操作报错 if value is None: self.cleaning_log.append(f"厚度为None，设为None") return None # 数值类型直接检查合理性 if isinstance(value, (int, float)): if value <= 0: # 厚度不可能为负 self.cleaning_log.append(f"厚度为负值({value})，设为None") return None return float(value) # 字符串需要更复杂的处理 if isinstance(value, str): value = value.strip().upper() # 常见的无效值枚举，避免误判 invalid_values = {'', 'N/A', 'NA', 'NONE', 'NULL', '-', '--', 'ERROR'} if value in invalid_values: self.cleaning_log.append(f"厚度为无效值({original})，设为None") return None # 移除单位（A、nm、um等），半导体常见单位 value = re.sub(r'[AaÅåµmnm]', '', value) value = value.strip() if not value: self.cleaning_log.append(f"厚度去除单位后为空({original})，设为None") return None try: result = float(value) if result <= 0: self.cleaning_log.append(f"厚度为负值({result})，设为None") return None return result except ValueError: self.cleaning_log.append(f"厚度无法转换({original})，设为None") return None return None def clean_lot_id(self, value) -> Optional[str]: """ 清洗Lot ID 为什么这样写： - MES中Lot ID前缀不统一（LOT_、LOT-、WAFER_等） - 统一转为大写，方便后续匹配和统计 """ if value is None: return None value = str(value).strip().upper() # 移除常见前缀，标准化格式 prefixes = ['LOT_', 'LOT-', 'LOT', 'WAFER_'] for prefix in prefixes: if value.startswith(prefix): value = value[len(prefix):] break # 只保留字母数字和连字符，移除其他特殊字符 value = re.sub(r'[^\w\-]', '', value) return value if value else None def clean_dataframe(self, df: pd.DataFrame, config: Optional[Dict] = None) -> pd.DataFrame: """ 清洗整个DataFrame 为什么这样写： - 使用配置字典，支持灵活扩展新的清洗规则 - 通过getattr动态调用方法，避免大量if-else - 记录清洗日志，便于审计和问题追溯 """ if config is None: # 默认配置：列名 -> 清洗方法名 config = { 'thickness': 'clean_thickness', 'lot_id': 'clean_lot_id', 'wafer_count': 'clean_wafer_count', 'date': 'clean_date', 'process': 'clean_process_name', } cleaned = df.copy() self.cleaning_log = [] for col, method_name in config.items(): if col in cleaned.columns: method = getattr(self, method_name) # 动态获取清洗方法 cleaned[col] = cleaned[col].apply(method) # 统计清洗效果 original_count = len(df) cleaned_count = cleaned.dropna(how='any').shape[0] removed_count = original_count - cleaned_count logger.info(f"清洗完成: {original_count}行 → {cleaned_count}行 (移除{removed_count}行)") logger.info(f"清洗日志: {len(self.cleaning_log)}条") return cleaned # 使用示例 if __name__ == '__main__': # 模拟脏数据 dirty_data = pd.DataFrame({ 'lot_id': ['FAB-001', 'LOT_FAB-002', 'fab-003', None], 'thickness': ['1250.5', 'N/A', -999, '1248.3A'], 'wafer_count': [25, '24', '0', 999], 'process': ['ETCH', 'etch', 'ETCH ', 'PHOTO'], 'date': ['2026/01/15', '2026-01-15', '2026.01.15', None], }) cleaner = MESDataCleaner() cleaned = cleaner.clean_dataframe(dirty_data) print(f"清洗完成，日志: {len(cleaner.cleaning_log)}条")

---

四、效果对比

维度 | 手动清洗 | 自动清洗 | 提升

|------|---------|---------|------|