# 数据工程：数据收集、清洗、标注与增强——为什么数据质量决定模型上限 **TL;DR：** 机器学习项目中，数据工程占据了60-80%的工作量，却决定了模型性能的理论上限。本文覆盖数据收集的策略设计、数据清洗的系统方法、数据标注的质量控制、数据增强的核心技术，帮助你建立一套完整的数据工程方法论。 ## 为什么这很重要 2016年，Google发表了一篇引起广泛讨论的论文《The unreasonable effectiveness of data》。核心观点很直接：在很多任务上，用更多、更好的数据训练一个简单模型，比用少量数据训练一个复杂模型效果更好。 这不是一句空话。后面一系列实验反复验证了这个结论： - 2021年，DeepMind的Chinchilla论文证明，训练数据量不够的时候，增大模型参数完全是浪费。70亿参数的模型在1.4万亿token上训练，比2800亿参数的模型在相同数据上训练效果更差，但当你把数据量增加到与参数匹配的程度，小模型反而更强。 - 2023年，LIMA论文用1000条精心筛选的高质量对话数据微调LLaMA-65B，在GPT-4评估中与GPT-4自身的回答质量相当。1000条数据打败了数十亿token的普通数据。 - 2024年，Phi系列模型（Microsoft）用"教科书质量"的合成数据训练，参数量只有7B，但在多个基准测试上超过了用海量网络数据训练的更大模型。 Andrew Ng在2021年提出了"Data-Centric AI"的概念：与其不断换模型架构，不如固定模型，集中精力提高数据质量。他把这个问题框架化了： > 如果你有一个固定的模型架构，改进数据质量的ROI远大于改进模型架构的ROI。 这背后的逻辑链是这样的： ``` 数据质量 → 决定模型上限（ceiling） 模型架构 → 决定逼近上限的速度和程度 训练技巧 → 决定实际能达到的水准 ``` 一条满是噪声的标注数据，就像给学生的教科书里写满了错误答案。不管这个学生有多聪明（模型有多大），学出来的知识都是有问题的。 更具体地说，数据质量的问题会在三个层面伤害你的模型： 1. **偏差（Bias）**：训练数据的分布与真实场景的分布不匹配，导致模型在特定群体或场景上表现差。人脸识别系统在深色皮肤上的表现明显差于浅色皮肤，根本原因就是训练数据中深色皮肤的样本太少。 2. **噪声（Noise）**：标注错误、数据损坏、不一致的标签，让模型学到错误的知识。ImageNet训练集中有约6%的标注错误，这些错误至今仍影响着基于ImageNet预训练的模型。 3. **覆盖不足（Coverage Gap）**：训练数据没有覆盖模型在真实场景中会遇到的所有情况。自动驾驶模型在晴天数据上训练得很好，但遇到暴雪就完全失效。 本文会从数据收集开始，经过清洗、标注、增强，完整地走一遍数据工程的全流程。每个环节都会提供具体的工具、方法和代码。 ## 核心概念 ### 数据的完整生命周期 一个ML项目的数据流经以下阶段： ``` 原始数据采集 → 数据探索与理解 → 数据清洗 → 数据标注 → 数据增强 → 特征工程 → 训练集/验证集/测试集划分 → 模型训练 → 数据监控 → 数据迭代 ``` 每个阶段都有可能引入问题，也都有机会提升数据质量。一个好的数据工程流水线应该： - **可追溯**：任何一条训练数据都能追溯到它的来源、处理过程、标注者 - **可复现**：给定相同的数据源和处理逻辑，能生成完全一致的训练数据 - **可迭代**：发现数据问题后，能快速修复并重新生成训练数据 - **可监控**：能持续检测数据质量的变化趋势 ### 数据质量的六个维度 Jim Gray在1998年提出了数据质量的经典框架，后来被ML社区广泛采用： 1. **准确性（Accuracy）**：数据值是否正确反映真实世界。标注"猫"的图片里确实是猫。 2. **完整性（Completeness）**：是否缺少必要的信息。图像数据缺少EXIF信息，用户行为数据缺少时间戳。 3. **一致性（Consistency）**：同一实体的不同记录是否一致。同一个用户在不同表中的年龄字段不一致。 4. **及时性（Timeliness）**：数据是否足够新。用2020年的新闻数据训练2026年的新闻推荐模型。 5. **唯一性（Uniqueness）**：是否存在重复记录。训练集中同一张图片出现多次，且标注不一致。 6. **有效性（Validity）**：数据是否符合定义的格式和范围。年龄字段出现负数或300。 在ML项目中，还要加一个维度： 7. **代表性（Representativeness）**：训练数据的分布是否代表模型在真实场景中会遇到的数据分布。 ### 数据工程的投入产出比 做一个粗略的估算。假设一个ML项目中： - 模型训练成本：GPU集群每小时100美元，训练100小时 = 10,000美元 - 数据标注成本：每条0.5美元，10万条 = 50,000美元 - 工程师时间成本：每月20,000美元，6个月 = 120,000美元 如果数据质量差导致模型表现不达标，需要重新训练： - 第二轮训练：10,000美元 - 重新标注：50,000美元 - 工程师调试时间：2个月 = 40,000美元 总额外成本：100,000美元。而这些成本中的大部分，如果在第一轮数据工程中就做好，是可以避免的。 这就是为什么数据工程值得你投入最大的精力。 ## 工作原理（简化的心智模型） ### 用做菜来理解数据工程 把训练ML模型想象成做一道菜： **数据收集** = 去菜市场买菜。你得知道要做什么菜（任务定义），才能决定买什么食材（数据类型）。买的时候要看食材新不新鲜（数据质量），品种对不对（数据分布），量够不够（数据量）。去一个靠谱的超市（可靠的数据源）比去路边摊（不可靠的数据源）安全得多。 **数据清洗** = 洗菜、切菜、挑出坏叶子。有的菜有泥巴（脏数据），得洗干净。有的菜有虫眼（异常值），得切掉。有的菜黄了（过期数据），得扔掉。如果懒得洗，直接下锅，菜里吃出沙子（模型输出垃圾），不能怪厨师（模型）。 **数据标注** = 给食材分类贴标签。这是牛肉，那是羊肉；这是有机蔬菜，那是普通蔬菜。标签贴错了（标注错误），做出来的菜味道就不对。如果请了多个人来分类，最好让他们独立操作，然后交叉检查（标注一致性）。 **数据增强** = 用有限的食材变出更多花样。你只有一颗白菜，但可以切丝、切片、切块、腌制、凉拌、炒着吃。每种处理方式都创造了不同的"体验"，虽然本质上还是白菜，但丰富了菜品的多样性。当然，你不能把白菜变成牛肉——增强不应该改变数据的本质语义。 **数据质量决定上限** = 不管你有多好的厨艺（模型架构）和多贵的灶台（GPU），如果食材本身不好，做出来的菜就是不好吃。顶级厨师用普通食材也能做出不错的菜，但普通食材的上限就是普通食材的上限。 这个类比的核心信息：**数据工程不是模型训练的前置步骤，它是模型训练的核心工作。** ## 工作原理（详细机制） ### 一、数据收集 数据收集是整个流水线的起点，也是最容易犯大错的地方。一个设计不当的数据收集策略，会导致后续所有的清洗和标注工作都无法弥补的偏差。 #### 1.1 数据源的选择 常见的数据源类型： | 数据源类型 | 优点 | 缺点 | 典型例子 | |-----------|------|------|---------| | 公开数据集 | 免费、标准化 | 可能与你的场景不匹配 | ImageNet, COCO, LAION-5B | | 网络爬取 | 大量、多样 | 版权问题、质量参差 | Common Crawl, YouTube字幕 | | 用户生成 | 真实、相关 | 隐私问题、需要脱敏 | 用户评论、上传图片 | | 合成数据 | 可控、无限量 | 分布可能与真实不同 | GPT生成的对话, Stable Diffusion生成的图像 | | 内部积累 | 高度相关 | 量可能不够 | 公司内部日志、历史交易 | | 传感器采集 | 精确、实时 | 设备成本、标注困难 | 自动驾驶车载摄像头数据 | 选择数据源时的关键考量： **覆盖度**：数据源是否覆盖了你模型需要处理的所有场景？如果你训练一个客服聊天机器人，只用投诉数据训练，它就学不会处理表扬和咨询。 **偏差**：数据源的采样过程是否有系统性偏差？推特数据偏向年轻人和英语用户，医院数据偏向已经就诊的患者（幸存者偏差）。 **时效性**：数据是否足够新？金融领域的数据贬值速度极快，三年前的股票走势数据对预测今天的走势几乎没有参考价值。 **合规性**：数据收集是否符合法律法规？GDPR（欧盟）、CCPA（加州）、《个人信息保护法》（中国）对数据收集都有严格要求。 #### 1.2 数据收集策略 **分层采样**（Stratified Sampling）是保证数据分布合理的核心方法。 假设你要收集一个情感分析数据集。如果你的数据来源是产品评论，正面评论可能占80%（因为满意的人不一定会留下评论，但不满意的人一定会——等等，实际情况可能相反：买东西满意的人也会留下好评）。不管实际比例是多少，如果你简单随机采样，得到的分布可能严重偏向某个类别。 分层采样的做法是：先确定你想要的类别比例（比如正:负:中性 = 1:1:1），然后从每个类别中分别采样。 ```python import numpy as np import pandas as pd from collections import Counter # 模拟一个不平衡的原始数据集 np.random.seed(42) n_samples = 10000 # 现实中的评论数据通常是不平衡的 # 5星好评最多，1星差评其次，2-4星相对少 ratings = np.random.choice( [1, 2, 3, 4, 5], size=n_samples, p=[0.15, 0.05, 0.10, 0.20, 0.50] # 5星占50% ) sentiments = [] for r in ratings: if r >= 4: sentiments.append('positive') elif r <= 2: sentiments.append('negative') else: sentiments.append('neutral') df = pd.DataFrame({ 'rating': ratings, 'text': [f'Product review text sample {i}' for i in range(n_samples)], 'sentiment': sentiments }) print("原始数据分布:") print(df['sentiment'].value_counts()) print(f"\n比例: {df['sentiment'].value_counts(normalize=True).round(3).to_dict()}") # 分层采样：每个类别采样相同数量 n_per_class = min(df['sentiment'].value_counts()) # 取最少的类别的数量 df_balanced = df.groupby('sentiment').apply( lambda x: x.sample(n=n_per_class, random_state=42), include_groups=False ).reset_index(drop=True) print(f"\n分层采样后数据分布:") print(df_balanced['sentiment'].value_counts()) ``` **主动学习**（Active Learning）是另一种重要的数据收集策略。核心思想是：不是所有数据对模型训练的价值都一样，让模型告诉你哪些数据最有标注价值。 ```python import numpy as np from sklearn.ensemble import RandomForestClassifier from sklearn.datasets import make_classification # 生成模拟数据 X_pool, y_pool = make_classification( n_samples=5000, n_features=20, n_informative=10, n_clusters_per_class=2, random_state=42 ) # 初始标注少量数据 n_initial = 50 indices = np.random.RandomState(42).permutation(len(X_pool)) X_train = X_pool[indices[:n_initial]] y_train = y_pool[indices[:n_initial]] # 未标注池 X_unlabeled = X_pool[indices[n_initial:]] y_unlabeled_true = y_pool[indices[n_initial:]] # 真实标签（模拟标注过程） # 主动学习循环 def uncertainty_sampling(classifier, X_unlabeled, n_query=10): """ 不确定性采样：选择模型最不确定的样本进行标注。 对于二分类，最不确定意味着预测概率最接近0.5。 """ probas = classifier.predict_proba(X_unlabeled) # 对于每个样本，取最大概率（最确定的类别的概率） # 最大概率越低 → 模型越不确定 → 越值得标注 uncertainty = 1 - np.max(probas, axis=1) query_indices = np.argsort(uncertainty)[-n_query:] return query_indices # 运行主动学习 clf = RandomForestClassifier(n_estimators=100, random_state=42) accuracies = [] query_sizes = [50, 100, 200, 300, 500, 800] for n_total in query_sizes: n_to_query = n_total - len(y_train) if n_to_query <= 0: continue # 不确定性采样 query_idx = uncertainty_sampling(clf, X_unlabeled, n_query=min(n_to_query, 20)) # "标注"被选中的样本（实际中需要人工标注） X_new = X_unlabeled[query_idx] y_new = y_unlabeled_true[query_idx] # 加入训练集 X_train = np.vstack([X_train, X_new]) y_train = np.concatenate([y_train, y_new]) # 从未标注池中移除 mask = np.ones(len(X_unlabeled), dtype=bool) mask[query_idx] = False X_unlabeled = X_unlabeled[mask] y_unlabeled_true = y_unlabeled_true[mask] # 重新训练 clf.fit(X_train, y_train) acc = clf.score(X_pool, y_pool) accuracies.append((len(y_train), acc)) print(f"训练集大小: {len(y_train):4d}, 准确率: {acc:.4f}") # 对比：随机采样 X_train_random = X_pool[indices[:n_initial]] y_train_random = y_pool[indices[:n_initial]] X_unlabeled_random = X_pool[indices[n_initial:]] y_unlabeled_random_true = y_pool[indices[n_initial:]] clf_random = RandomForestClassifier(n_estimators=100, random_state=42) random_accuracies = [] for n_total in query_sizes: n_to_query = n_total - len(y_train_random) if n_to_query <= 0: continue # 随机采样 query_idx = np.random.RandomState(42).choice( len(X_unlabeled_random), size=min(n_to_query, 20), replace=False ) X_new = X_unlabeled_random[query_idx] y_new = y_unlabeled_random_true[query_idx] X_train_random = np.vstack([X_train_random, X_new]) y_train_random = np.concatenate([y_train_random, y_new]) mask = np.ones(len(X_unlabeled_random), dtype=bool) mask[query_idx] = False X_unlabeled_random = X_unlabeled_random[mask] y_unlabeled_random_true = y_unlabeled_random_true[mask] clf_random.fit(X_train_random, y_train_random) acc = clf_random.score(X_pool, y_pool) random_accuracies.append((len(y_train_random), acc)) print(f"[随机采样] 训练集大小: {len(y_train_random):4d}, 准确率: {acc:.4f}") ``` #### 1.3 合成数据生成 2024年以来，用大模型生成合成训练数据成为一个重要趋势。关键在于质量控制： ```python """ 使用LLM生成合成对话数据的框架 这里展示质量控制的核心逻辑，不依赖具体的LLM API """ import random from dataclasses import dataclass, field from typing import List, Optional from collections import Counter @dataclass class SyntheticDataConfig: """合成数据生成配置""" target_count: int = 1000 categories: List[str] = field(default_factory=lambda: ['greeting', 'question', 'complaint', 'feedback', 'request']) quality_threshold: float = 0.8 # 质量分数阈值 max_duplicates_ratio: float = 0.05 # 最大重复率 diversity_target: float = 0.9 # 多样性目标 @dataclass class SyntheticSample: """一条合成数据""" text: str category: str quality_score: float source_prompt: str generation_params: dict class SyntheticDataPipeline: """合成数据生成与质量控制管线""" def __init__(self, config: SyntheticDataConfig): self.config = config self.generated: List[SyntheticSample] = [] self.rejected: List[SyntheticSample] = [] def generate_batch(self, prompts: List[str], category: str) -> List[SyntheticSample]: """ 模拟批量生成合成数据 实际使用时会调用GPT-4/Claude等LLM """ samples = [] for prompt in prompts: # 模拟LLM生成 quality_score = random.uniform(0.5, 1.0) sample = SyntheticSample( text=f"Generated response for: {prompt}", category=category, quality_score=quality_score, source_prompt=prompt, generation_params={'temperature': 0.7, 'model': 'gpt-4'} ) if quality_score >= self.config.quality_threshold: self.generated.append(sample) samples.append(sample) else: self.rejected.append(sample) return samples def check_quality(self) -> dict: """质量检查报告""" if not self.generated: return {"status": "no_data"} # 1. 类别分布 category_dist = Counter(s.category for s in self.generated) # 2. 质量分数统计 quality_scores = [s.quality_score for s in self.generated] # 3. 重复率检查（简化版：基于文本前缀） seen_prefixes = set() duplicates = 0 for s in self.generated: prefix = s.text[:50] if prefix in seen_prefixes: duplicates += 1 seen_prefixes.add(prefix) duplicate_ratio = duplicates / len(self.generated) if self.generated else 0 report = { "total_generated": len(self.generated), "total_rejected": len(self.rejected), "rejection_rate": len(self.rejected) / (len(self.generated) + len(self.rejected)), "category_distribution": dict(category_dist), "avg_quality_score": sum(quality_scores) / len(quality_scores), "min_quality_score": min(quality_scores), "duplicate_ratio": duplicate_ratio, "quality_passed": duplicate_ratio <= self.config.max_duplicates_ratio } return report # 使用示例 config = SyntheticDataConfig(target_count=100) pipeline = SyntheticDataPipeline(config) # 为每个类别生成数据 for category in config.categories: prompts = [f"Write a {category} message about topic {i}" for i in range(25)] pipeline.generate_batch(prompts, category) report = pipeline.check_quality() print("合成数据质量报告:") for k, v in report.items(): print(f" {k}: {v}") ``` ### 二、数据清洗 数据清洗是最不性感但最重要的工作。经验法则：**花在数据清洗上的每一小时，可以为模型训练节省十小时。** #### 2.1 常见的数据质量问题 ```python import pandas as pd import numpy as np from datetime import datetime, timedelta # 创建一个模拟的真实数据集，故意包含各种问题 np.random.seed(42) n = 5000 def create_dirty_dataset(n=5000): """创建一个包含各种数据质量问题的数据集""" # 基础数据 ages = np.random.normal(35, 12, n).astype(int) ages = np.clip(ages, 0, 100) salaries = np.random.lognormal(10.5, 0.6, n).astype(int) emails = [f"user{i}@{'gmail' if i % 3 != 0 else 'yahoo'}.com" for i in range(n)] departments = np.random.choice( ['Engineering', 'Marketing', 'Sales', 'HR', 'Finance'], size=n, p=[0.35, 0.20, 0.20, 0.10, 0.15] ) join_dates = [ datetime(2020, 1, 1) + timedelta(days=int(x)) for x in np.random.uniform(0, 1825, n) # 2020-2025 ] # === 注入数据质量问题 === # 问题1: 缺失值 (约5%) age_mask = np.random.random(n) < 0.05 ages[age_mask] = np.nan # 注意：整数数组不能直接赋nan salary_mask = np.random.random(n) < 0.08 salaries[salary_mask] = -1 # 用-1表示缺失（常见的不良实践） # 问题2: 异常值 ages[10] = 200 # 不合理的年龄 ages[20] = -5 # 负年龄 salaries[50] = 1 # 异常低的薪资 salaries[100] = 99999999 # 异常高的薪资 # 问题3: 重复记录 duplicate_indices = np.random.choice(n, size=100, replace=False) # 问题4: 格式不一致 dept_clean = list(departments) for i in np.random.choice(n, size=50, replace=False): dept_clean[i] = dept_clean[i].lower() # 有的小写 for i in np.random.choice(n, size=30, replace=False): dept_clean[i] = dept_clean[i].strip() + ' ' # 多余空格 dept_clean[5] = 'Eng' # 缩写 dept_clean[15] = 'eng' # 小写缩写 # 问题5: 数据类型问题 emails[30] = 'invalid-email' emails[60] = 'user@' # 不完整的邮箱 # 构建DataFrame data = pd.DataFrame({ 'age': ages.astype(float), 'salary': salaries, 'email': emails, 'department': dept_clean, 'join_date': join_dates }) # 添加重复行 duplicates = data.iloc[duplicate_indices].copy() data = pd.concat([data, duplicates], ignore_index=True) return data df = create_dirty_dataset() print(f"数据集大小: {df.shape}") print(f"\n前5行:") print(df.head()) ``` #### 2.2 系统化的数据清洗流水线 ```python from typing import Tuple, List, Dict, Any import re class DataCleaner: """ 系统化的数据清洗流水线 每个清洗步骤都记录变更，支持审计和回滚 """ def __init__(self, df: pd.DataFrame): self.original = df.copy() self.df = df.copy() self.cleaning_log = [] self.stats = { 'original_rows': len(df), 'original_cols': len(df.columns), } def log_action(self, step: str, affected: int, detail: str): self.cleaning_log.append({ 'step': step, 'affected_rows': affected, 'detail': detail, 'remaining_rows': len(self.df) }) def remove_duplicates(self, subset: List[str] = None) -> 'DataCleaner': """去除重复记录""" before = len(self.df) self.df = self.df.drop_duplicates(subset=subset, keep='first') removed = before - len(self.df) self.log_action('remove_duplicates', removed, f'基于 {subset or "所有列"} 去重') return self def handle_missing_values(self, strategy: Dict[str, str]) -> 'DataCleaner': """ 处理缺失值 strategy: {列名: 'drop'|'mean'|'median'|'mode'|'fill_value'} """ for col, method in strategy.items(): if col not in self.df.columns: continue missing_count = self.df[col].isna().sum() if method == 'drop': before = len(self.df) self.df = self.df.dropna(subset=[col]) self.log_action('handle_missing', before - len(self.df), f'删除 {col} 缺失行') elif method == 'mean': fill_val = self.df[col].mean() self.df[col] = self.df[col].fillna(fill_val) self.log_action('handle_missing', missing_count, f'{col} 用均值 {fill_val:.2f} 填充') elif method == 'median': fill_val = self.df[col].median() self.df[col] = self.df[col].fillna(fill_val) self.log_action('handle_missing', missing_count, f'{col} 用中位数 {fill_val:.2f} 填充') elif method == 'mode': fill_val = self.df[col].mode()[0] self.df[col] = self.df[col].fillna(fill_val) self.log_action('handle_missing', missing_count, f'{col} 用众数 {fill_val} 填充') elif isinstance(method, (int, float, str)): self.df[col] = self.df[col].fillna(method) self.log_action('handle_missing', missing_count, f'{col} 用 {method} 填充') return self def remove_outliers_iqr(self, columns: List[str], factor: float = 1.5) -> 'DataCleaner': """ 使用IQR方法去除异常值 factor: IQR倍数，1.5是标准值，3.0更保守 """ before = len(self.df) for col in columns: Q1 = self.df[col].quantile(0.25) Q3 = self.df[col].quantile(0.75) IQR = Q3 - Q1 lower = Q1 - factor * IQR upper = Q3 + factor * IQR mask = (self.df[col] >= lower) & (self.df[col] <= upper) removed = (~mask).sum() self.df = self.df[mask] self.log_action('remove_outliers', removed, f'{col}: 移除超出 [{lower:.1f}, {upper:.1f}] 的值') return self def remove_outliers_zscore(self, columns: List[str], threshold: float = 3.0) -> 'DataCleaner': """ 使用Z-score方法去除异常值 threshold: Z-score阈值，通常用3.0 """ before = len(self.df) for col in columns: mean = self.df[col].mean() std = self.df[col].std() if std == 0: continue z_scores = np.abs((self.df[col] - mean) / std) mask = z_scores < threshold removed = (~mask).sum() self.df = self.df[mask] self.log_action('remove_outliers', removed, f'{col}: 移除Z-score > {threshold} 的值') return self def standardize_categories(self, column: str, mapping: Dict[str, str]) -> 'DataCleaner': """ 标准化分类变量 mapping: {原始值: 标准值} """ before_values = set(self.df[column].unique()) # 先去除前后空格 self.df[column] = self.df[column].str.strip() # 应用映射 self.df[column] = self.df[column].replace(mapping) after_values = set(self.df[column].unique()) new_categories = after_values - before_values self.log_action('standardize_categories', 0, f'{column}: {len(mapping)} 个值被标准化') return self def validate_and_fix_types(self, type_spec: Dict[str, str]) -> 'DataCleaner': """ 验证和修复数据类型 type_spec: {列名: 'int'|'float'|'datetime'|'string'} """ for col, dtype in type_spec.items(): if col not in self.df.columns: continue try: if dtype == 'int': self.df[col] = pd.to_numeric(self.df[col], errors='coerce').astype('Int64') elif dtype == 'float': self.df[col] = pd.to_numeric(self.df[col], errors='coerce') elif dtype == 'datetime': self.df[col] = pd.to_datetime(self.df[col], errors='coerce') elif dtype == 'string': self.df[col] = self.df[col].astype(str) self.log_action('validate_types', 0, f'{col} 转换为 {dtype}') except Exception as e: self.log_action('validate_types', 0, f'{col} 类型转换失败: {e}') return self def get_report(self) -> dict: """生成清洗报告""" return { 'original_shape': self.original.shape, 'final_shape': self.df.shape, 'rows_removed': len(self.original) - len(self.df), 'removal_rate': (len(self.original) - len(self.df)) / len(self.original), 'steps': len(self.cleaning_log), 'step_details': self.cleaning_log } # 使用清洗流水线 df = create_dirty_dataset() # 定义部门名称映射 dept_mapping = { 'engineering': 'Engineering', 'marketing': 'Marketing', 'sales': 'Sales', 'hr': 'HR', 'finance': 'Finance', 'Eng': 'Engineering', 'eng': 'Engineering', } cleaner = DataCleaner(df) result = (cleaner .remove_duplicates() .validate_and_fix_types({'age': 'float', 'salary': 'int'}) .handle_missing_values({'age': 'median', 'salary': 'median'}) .remove_outliers_iqr(['age', 'salary'], factor=1.5) .standardize_categories('department', dept_mapping) ) report = cleaner.get_report() print(f"\n清洗报告:") print(f" 原始数据: {report['original_shape']}") print(f" 清洗后: {report['final_shape']}") print(f" 移除率: {report['removal_rate']:.2%}") print(f" 清洗步骤: {report['steps']}") print(f"\n详细步骤:") for step in report['step_details']: print(f" [{step['step']}] 影响 {step['affected_rows']} 行 - {step['detail']}") ``` #### 2.3 数据质量评估框架 清洗完数据后，你还需要一个系统化的方法来评估清洗后的数据质量： ```python class DataQualityAssessor: """ 数据质量评估器 从六个维度评估数据质量 """ def __init__(self, df: pd.DataFrame, label_column: str = None): self.df = df self.label_col = label_column def assess_accuracy(self, rules: Dict[str, tuple]) -> dict: """ 准确性评估 rules: {列名: (最小值, 最大值)} """ results = {} for col, (min_val, max_val) in rules.items(): if col not in self.df.columns: continue valid_mask = (self.df[col] >= min_val) & (self.df[col] <= max_val) accuracy = valid_mask.mean() results[col] = { 'accuracy': accuracy, 'out_of_range': (~valid_mask).sum(), 'range': (min_val, max_val) } return results def assess_completeness(self) -> dict: """完整性评估""" total_cells = self.df.shape[0] * self.df.shape[1] missing_cells = self.df.isna().sum().sum() results = { 'overall_completeness': 1 - missing_cells / total_cells, 'columns': {} } for col in self.df.columns: missing = self.df[col].isna().sum() results['columns'][col] = { 'completeness': 1 - missing / len(self.df), 'missing_count': missing } return results def assess_consistency(self, consistency_rules: Dict[str, callable]) -> dict: """ 一致性评估 consistency_rules: {规则名: 判断函数(返回bool Series)} """ results = {} for rule_name, rule_func in consistency_rules.items(): consistent = rule_func(self.df) results[rule_name] = { 'consistency': consistent.mean(), 'inconsistent_count': (~consistent).sum() } return results def assess_uniqueness(self, subset: List[str] = None) -> dict: """唯一性评估""" total = len(self.df) duplicates = self.df.duplicated(subset=subset).sum() return { 'uniqueness': 1 - duplicates / total, 'duplicate_count': duplicates, 'total_records': total } def assess_representativeness(self, expected_dist: Dict[str, float] = None) -> dict: """ 代表性评估 比较实际分布与期望分布的差异 """ if self.label_col is None or expected_dist is None: return {'status': '需要指定标签列和期望分布'} actual_dist = self.df[self.label_col].value_counts(normalize=True).to_dict() # 计算KL散度 kl_divergence = 0 for category, expected_p in expected_dist.items(): actual_p = actual_dist.get(category, 1e-10) kl_divergence += expected_p * np.log(expected_p / max(actual_p, 1e-10)) return { 'actual_distribution': actual_dist, 'expected_distribution': expected_dist, 'kl_divergence': kl_divergence, 'interpretation': 'KL散度越小，代表性越好。0表示完全匹配。' } def full_report(self, accuracy_rules=None, consistency_rules=None, uniqueness_subset=None, expected_dist=None) -> dict: """生成完整质量报告""" report = { 'dataset_shape': self.df.shape, 'completeness': self.assess_completeness(), 'uniqueness': self.assess_uniqueness(uniqueness_subset), } if accuracy_rules: report['accuracy'] = self.assess_accuracy(accuracy_rules) if consistency_rules: report['consistency'] = self.assess_consistency(consistency_rules) if expected_dist: report['representativeness'] = self.assess_representativeness(expected_dist) return report # 使用示例 df = create_dirty_dataset() assessor = DataQualityAssessor(df, label_column='department') accuracy_rules = { 'age': (0, 120), 'salary': (0, 1000000), } consistency_rules = { 'salary_positive': lambda df: df['salary'] > 0, 'valid_age_range': lambda df: (df['age'] >= 18) & (df['age'] <= 65), } expected_dist = { 'Engineering': 0.35, 'Marketing': 0.20, 'Sales': 0.20, 'HR': 0.10, 'Finance': 0.15, } report = assessor.full_report( accuracy_rules=accuracy_rules, consistency_rules=consistency_rules, expected_dist=expected_dist ) import json print(json.dumps(report, indent=2, default=str, ensure_ascii=False)) ``` ### 三、数据标注 数据标注是将原始数据转化为模型可学习的信号的过程。标注质量直接决定了监督学习的天花板。 #### 3.1 标注方法论 **标注指南编写**：这是最容易被忽视但影响最大的环节。一份好的标注指南应该： 1. 包含每种标签的精确定义，不能有歧义 2. 包含边界情况的判断规则 3. 包含标注示例和反面案例 4. 标注者可以在不求助的情况下独立完成标注 ```python # 标注指南示例（情感分析任务） ANNOTATION_GUIDE = """ # 情感分析标注指南 v2.3 ## 标签定义 ### 正面 (positive) 文本明确表达了满意、赞赏、推荐等正面情感。 - 包含明确的赞美词汇（"很好"、"推荐"、"满意"） - 或者整体语义明显正面 ### 负面 (negative) 文本明确表达了不满、批评、抱怨等负面情感。 - 包含明确的负面词汇（"差"、"失望"、"不推荐"） - 或者整体语义明显负面 ### 中性 (neutral) 以下情况标注为中性： 1. 纯事实陈述，不含情感色彩（"商品周一送达"） 2. 正面和负面情感混合，难以判断整体倾向（"质量好但价格贵"） 3. 信息不足，无法判断（"收到了"） ## 边界情况 | 情况 | 标注 | 原因 | |------|------|------| | "还行吧" | neutral | 口语中的"还行"偏向中性 | | "一般般，不功不过" | neutral | 明确的中间态度 | | "质量可以，但物流太慢了" | negative | 当负面评价涉及核心体验时，整体偏负面 | | "虽然贵了点，但物有所值" | positive | "物有所值"是明确的正面评价 | | 讽刺语气（"真是太'好'了"） | negative | 按实际语义而非字面标注 | ## 标注示例 正面: "这个产品质量超乎预期，强烈推荐给大家！" 负面: "用了两天就坏了，浪费钱。" 中性: "订单已确认，预计3天内送达。" """ ``` #### 3.2 标注质量控制 标注质量控制的核心是**一致性度量**。多个标注者对同一条数据的标注越一致，说明标注质量越高。 ```python from typing import List, Dict from itertools import combinations class AnnotationQualityControl: """标注质量控制工具""" def __init__(self, annotations: Dict[str, Dict[str, str]]): """ annotations: { 'sample_001': {'annotator_A': 'positive', 'annotator_B': 'positive', 'annotator_C': 'neutral'}, ... } """ self.annotations = annotations def cohen_kappa(self, annotator1: str, annotator2: str) -> float: """ Cohen's Kappa 系数 衡量两个标注者之间的一致性，排除了偶然一致的可能性 kappa > 0.8: 优秀 0.6 < kappa <= 0.8: 良好 0.4 < kappa <= 0.6: 中等 kappa <= 0.4: 差，需要重新审查标注指南 """ # 获取两个标注者共同标注的样本 common_samples = [ sid for sid in self.annotations if annotator1 in self.annotations[sid] and annotator2 in self.annotations[sid] ] if not common_samples: return 0.0 # 统计一致性 all_labels = set() for sid in common_samples: all_labels.add(self.annotations[sid][annotator1]) all_labels.add(self.annotations[sid][annotator2]) labels = sorted(all_labels) n_labels = len(labels) label_to_idx = {l: i for i, l in enumerate(labels)} # 构建混淆矩阵 confusion = np.zeros((n_labels, n_labels)) for sid in common_samples: i = label_to_idx[self.annotations[sid][annotator1]] j = label_to_idx[self.annotations[sid][annotator2]] confusion[i][j] += 1 n = len(common_samples) # 观察到的一致性 p_observed = np.trace(confusion) / n # 期望的偶然一致性 p_expected = sum( (confusion[i].sum() / n) * (confusion[:, j].sum() / n) for i in range(n_labels) for j in range(n_labels) if i == j ) if p_expected == 1.0: return 1.0 kappa = (p_observed - p_expected) / (1 - p_expected) return kappa def fleiss_kappa(self) -> float: """ Fleiss' Kappa 系数 衡量多个标注者之间的一致性 适用于每个样本由不同组的标注者标注的情况 """ # 统计每个样本的标注分布 all_labels = set() for sid in self.annotations: for ann in self.annotations[sid].values(): all_labels.add(ann) labels = sorted(all_labels) n_labels = len(labels) label_to_idx = {l: i for i, l in enumerate(labels)} n_samples = len(self.annotations) # 构建标注矩阵 (n_samples x n_labels) # 每个单元格表示该样本被标注为该类别的次数 matrix = np.zeros((n_samples, n_labels)) for i, sid in enumerate(self.annotations): for ann_label in self.annotations[sid].values(): matrix[i][label_to_idx[ann_label]] += 1 # 每个样本的标注者数量 n_annotations_per_sample = matrix.sum(axis=1) n = n_annotations_per_sample.mean() # 平均标注者数 # P_i: 每个样本的一致性 P = (1 / (n * (n - 1))) * ( np.sum(matrix ** 2, axis=1) - n_annotations_per_sample ) P_bar = P.mean() # p_j: 每个类别的比例 p = matrix.sum(axis=0) / matrix.sum() # P_e: 期望一致性 P_e = np.sum(p ** 2) if P_e == 1.0: return 1.0 kappa = (P_bar - P_e) / (1 - P_e) return kappa def disagreement_analysis(self) -> List[dict]: """分析标注不一致的样本，优先审查""" disagreements = [] for sid, anns in self.annotations.items(): labels = list(anns.values()) unique_labels = set(labels) if len(unique_labels) > 1: # 计算不一致程度 from collections import Counter label_counts = Counter(labels) majority_label, majority_count = label_counts.most_common(1)[0] agreement_rate = majority_count / len(labels) disagreements.append({ 'sample_id': sid, 'annotations': anns, 'agreement_rate': agreement_rate, 'majority_label': majority_label, 'priority': 'high' if agreement_rate < 0.5 else 'medium' }) # 按一致性从低到高排序 disagreements.sort(key=lambda x: x['agreement_rate']) return disagreements # 模拟标注数据 annotations = {} np.random.seed(42) labels = ['positive', 'negative', 'neutral'] sample_ids = [f'sample_{i:04d}' for i in range(500)] for sid in sample_ids: true_label = np.random.choice(labels, p=[0.4, 0.3, 0.3]) # 模拟三个标注者，一致性大约80-90% ann = {} for annotator in ['A', 'B', 'C']: if np.random.random() < 0.85: ann[annotator] = true_label else: other_labels = [l for l in labels if l != true_label] ann[annotator] = np.random.choice(other_labels) annotations[sid] = ann qc = AnnotationQualityControl(annotations) # 计算两两一致性 print("Cohen's Kappa:") for a1, a2 in combinations(['A', 'B', 'C'], 2): kappa = qc.cohen_kappa(a1, a2) print(f" {a1} vs {a2}: {kappa:.4f}") # 计算整体一致性 fk = qc.fleiss_kappa() print(f"\nFleiss' Kappa: {fk:.4f}") # 分析不一致样本 disagreements = qc.disagreement_analysis() print(f"\n不一致样本数: {len(disagreements)}") print(f"高优先级（一致性<50%）: {sum(1 for d in disagreements if d['priority'] == 'high')}") if disagreements: print(f"\n最不一致的5个样本:") for d in disagreements[:5]: print(f" {d['sample_id']}: {d['annotations']} (一致性: {d['agreement_rate']:.2f})") ``` #### 3.3 标注工具生态 主流标注工具对比： | 工具 | 适用场景 | 特点 | 价格 | |------|---------|------|------| | Label Studio | 通用 | 开源、可扩展、支持多模态 | 免费开源 | | Labelbox | 企业级 | 强大的QA流程、团队协作 | 按量计费 | | Scale AI | 大规模 | 人工标注服务+平台 | 按条计费 | | Roboflow | 计算机视觉 | CV全流程、模型辅助标注 | 免费层+付费 | | Prodigy | NLP | spaCy生态、主动学习 | 买断制 | | CVAT | 计算机视觉 | Intel开源、视频标注 | 免费开源 | | Cleanlab Studio | 自动化 | 自动发现标注错误 | 免费层+付费 | ### 四、数据增强 数据增强是在不收集新数据的情况下，从现有数据中创造更多训练样本的技术。它是提升模型泛化能力的最廉价手段。 #### 4.1 图像增强 图像增强是最成熟的领域，有大量经过验证的方法： ```python import numpy as np from typing import List, Tuple, Callable # ============================================================ # 图像增强方法实现 # 为了可读性，使用numpy手动实现，不依赖专门的图像库 # ============================================================ class ImageAugmentor: """图像增强工具集""" @staticmethod def random_flip(image: np.ndarray, p: float = 0.5) -> np.ndarray: """随机水平翻转""" if np.random.random() < p: return image[:, ::-1, :] return image @staticmethod def random_rotation(image: np.ndarray, max_angle: float = 15) -> np.ndarray: """ 随机旋转（简化版） 实际中应使用scipy.ndimage.rotate或OpenCV """ angle = np.random.uniform(-max_angle, max_angle) # 简化实现：用transpose模拟90度倍数旋转 if abs(angle) > 45: return np.transpose(image, (1, 0, 2)) return image @staticmethod def random_crop(image: np.ndarray, crop_ratio: float = 0.9) -> np.ndarray: """ 随机裁剪 crop_ratio: 裁剪区域占原图的比例 """ h, w = image.shape[:2] new_h, new_w = int(h * crop_ratio), int(w * crop_ratio) top = np.random.randint(0, h - new_h + 1) left = np.random.randint(0, w - new_w + 1) cropped = image[top:top+new_h, left:left+new_w] # 调整回原始尺寸（简化版：用repeat近似） scale_h = h // new_h scale_w = w // new_w if scale_h > 0 and scale_w > 0: return np.repeat(np.repeat(cropped, scale_h, axis=0), scale_w, axis=1)[:h, :w] return cropped @staticmethod def color_jitter(image: np.ndarray, brightness: float = 0.2, contrast: float = 0.2, saturation: float = 0.2) -> np.ndarray: """ 颜色抖动 brightness, contrast, saturation: 变化幅度 """ result = image.copy().astype(np.float32) # 亮度 if brightness > 0: factor = 1.0 + np.random.uniform(-brightness, brightness) result *= factor # 对比度 if contrast > 0: factor = 1.0 + np.random.uniform(-contrast, contrast) mean = result.mean() result = mean + factor * (result - mean) # 饱和度（简化版：调整绿色通道） if saturation > 0: factor = 1.0 + np.random.uniform(-saturation, saturation) gray = result.mean(axis=2, keepdims=True) result = gray + factor * (result - gray) return np.clip(result, 0, 255).astype(np.uint8) @staticmethod def cutout(image: np.ndarray, n_holes: int = 1, hole_size: int = 16) -> np.ndarray: """ Cutout: 随机遮挡图像的区域 迫使模型不只关注局部特征 """ result = image.copy() h, w = image.shape[:2] for _ in range(n_holes): y = np.random.randint(0, h) x = np.random.randint(0, w) y1 = max(0, y - hole_size // 2) y2 = min(h, y + hole_size // 2) x1 = max(0, x - hole_size // 2) x2 = min(w, x + hole_size // 2) result[y1:y2, x1:x2] = 0 return result @staticmethod def mixup(image1: np.ndarray, image2: np.ndarray, label1: int, label2: int, alpha: float = 0.4) -> Tuple[np.ndarray, Tuple[float, float]]: """ Mixup: 混合两张图片和它们的标签 论文: mixup: Beyond Empirical Risk Minimization (2018) alpha: Dirichlet分布参数，越小混合越轻微 返回: (混合后的图片, (标签1的权重, 标签2的权重)) """ lam = np.random.beta(alpha, alpha) mixed_image = (lam * image1 + (1 - lam) * image2).astype(np.uint8) return mixed_image, (lam, 1 - lam) @staticmethod def cutmix(image1: np.ndarray, image2: np.ndarray, label1: int, label2: int, alpha: float = 1.0) -> Tuple[np.ndarray, Tuple[float, float]]: """ CutMix: 从一张图裁剪区域粘贴到另一张图 论文: CutMix: Regularization Strategy (2019) """ h, w = image1.shape[:2] lam = np.random.beta(alpha, alpha) # 计算裁剪区域 cut_ratio = np.sqrt(1 - lam) cut_w = int(w * cut_ratio) cut_h = int(h * cut_ratio) cx = np.random.randint(0, w) cy = np.random.randint(0, h) x1 = max(0, cx - cut_w // 2) y1 = max(0, cy - cut_h // 2) x2 = min(w, cx + cut_w // 2) y2 = min(h, cy + cut_h // 2) result = image1.copy() result[y1:y2, x1:x2] = image2[y1:y2, x1:x2] # 重新计算lambda（基于实际裁剪面积） actual_lam = 1 - (x2 - x1) * (y2 - y1) / (w * h) return result, (actual_lam, 1 - actual_lam) # 演示所有增强方法 augmentor = ImageAugmentor() # 创建模拟图像 (224x224x3) fake_image = np.random.randint(0, 256, (224, 224, 3), dtype=np.uint8) fake_image2 = np.random.randint(0, 256, (224, 224, 3), dtype=np.uint8) print("图像增强方法演示:") print(f" 原始图像: {fake_image.shape}") flipped = augmentor.random_flip(fake_image) print(f" 水平翻转: {flipped.shape}") cropped = augmentor.random_crop(fake_image, crop_ratio=0.8) print(f" 随机裁剪: {cropped.shape}") jittered = augmentor.color_jitter(fake_image) print(f" 颜色抖动: {jittered.shape}, dtype: {jittered.dtype}") cutout_img = augmentor.cutout(fake_image, hole_size=32) print(f" Cutout: {cutout_img.shape}") mixed, (w1, w2) = augmentor.mixup(fake_image, fake_image2, 0, 1) print(f" Mixup: {mixed.shape}, 标签权重: ({w1:.3f}, {w2:.3f})") cutmixed, (w1, w2) = augmentor.cutmix(fake_image, fake_image2, 0, 1) print(f" CutMix: {cutmixed.shape}, 标签权重: ({w1:.3f}, {w2:.3f})") ``` #### 4.2 文本增强 文本增强比图像增强更具挑战性，因为文本的语义更容易被改变。 ```python import random from typing import List class TextAugmentor: """文本增强方法""" @staticmethod def synonym_replacement(text: str, replacement_ratio: float = 0.1) -> str: """ 同义词替换 将文本中的部分词替换为同义词 """ # 简化的同义词词典（实际中应使用WordNet或专业词典） synonyms = { '好': ['不错', '优秀', '棒'], '差': ['不好', '糟糕', '劣质'], '快': ['迅速', '快捷', '敏捷'], '慢': ['迟缓', '缓慢', '磨蹭'], '大': ['巨大', '庞大', '硕大'], 'good': ['great', 'excellent', 'fine', 'nice'], 'bad': ['poor', 'terrible', 'awful'], 'fast': ['quick', 'rapid', 'swift'], 'slow': ['gradual', 'sluggish'], } words = text.split() n_replace = max(1, int(len(words) * replacement_ratio)) indices = list(range(len(words))) random.shuffle(indices) replaced = 0 for idx in indices: word = words[idx].lower() if word in synonyms: words[idx] = random.choice(synonyms[word]) replaced += 1 if replaced >= n_replace: break return ' '.join(words) @staticmethod def random_deletion(text: str, deletion_prob: float = 0.1) -> str: """ 随机删除 以一定概率删除文本中的词 """ words = text.split() if len(words) == 1: return text remaining = [w for w in words if random.random() > deletion_prob] if not remaining: # 至少保留一个词 return words[0] return ' '.join(remaining) @staticmethod def random_swap(text: str, n_times: int = 1) -> str: """ 随机交换 随机交换文本中两个词的位置 """ words = text.split() if len(words) < 2: return text for _ in range(n_times): i, j = random.sample(range(len(words)), 2) words[i], words[j] = words[j], words[i] return ' '.join(words) @staticmethod def back_translation(text: str) -> str: """ 回译（Back Translation） 原文 → 翻译到另一种语言 → 翻译回原文 产生语义相同但表达不同的文本 实际使用需要调用翻译API（Google Translate, DeepL等） 这里展示框架逻辑 """ # 模拟回译过程 # 实际流程: text -> translate(text, src='zh', tgt='en') -> translate(result, src='en', tgt='zh') simulated_translations = { '这个产品质量很好': ['这款商品的品质相当出色', '这件产品的质量挺不错的'], '物流太慢了': ['配送速度太慢了', '送货太迟了', '发货太慢了'], } if text in simulated_translations: return random.choice(simulated_translations[text]) return text @staticmethod def contextual_insertion(text: str, insert_prob: float = 0.1) -> str: """ 上下文插入 使用语言模型在合适的位置插入词 简化版：在随机位置插入连接词或修饰词 """ connectors = ['而且', '实际上', '确实', '说实话', '不过'] words = list(text) # 中文按字符处理 result = [] for char in words: result.append(char) if random.random() < insert_prob and char in '，。！？；': result.append(random.choice(connectors)) return ''.join(result) # 演示文本增强 text_augmentor = TextAugmentor() sample_texts = [ "这个 产品 质量 很好 而且 发货 速度 fast", "这部 电影 bad 但是 演员 表演 good", "服务 fast 价格 cheap 总体 体验 good" ] print("文本增强演示:") for text in sample_texts: print(f"\n原文: {text}") print(f" 同义词替换: {text_augmentor.synonym_replacement(text)}") print(f" 随机删除: {text_augmentor.random_deletion(text)}") print(f" 随机交换: {text_augmentor.random_swap(text)}") ``` #### 4.3 增强策略的选择 不同的任务和数据量需要不同的增强策略： | 场景 | 推荐增强方法 | 注意事项 | |------|-------------|---------| | 小数据集CV分类 | Mixup, CutMix, RandAugment | 增强强度可以大一些 | | 目标检测 | 随机裁剪（保持bbox）, Mosaic | 不应该改变bbox的语义 | | NLP分类 | 回译, 同义词替换 | 避免改变句子情感 | | NER | 同义词替换, 实体替换 | 必须同步更新标签 | | 语音识别 | 速度扰动, SpecAugment | 不能改变语音内容 | | 大数据集 | 轻度增强或不增强 | 数据量够大时增强的收益递减 | 一个常见的错误是过度增强，导致增强后的数据与真实分布偏差太大。CutMix把猫的头接到狗的身上，这对ImageNet分类来说有正则化效果，但如果你训练的是医学图像分析模型，这种增强可能引入不合法的图像。 **增强策略验证方法**：先用增强后的数据训练一个小模型，在验证集上评估。如果增强后性能提升，继续；如果下降，减小增强强度或更换方法。 ```python def validate_augmentation_strategy( train_func, # 训练函数 evaluate_func, # 评估函数 train_data, # 原始训练数据 augmentation_funcs: List[Callable], # 候选增强方法 n_trials: int = 3 # 每种方法重复试验次数 ) -> dict: """ 验证增强策略的有效性 核心思路：对比有增强和无增强的训练效果 如果增强后效果更好，说明增强策略有效 """ results = {'baseline': [], 'augmented': {}} # 基线：不使用增强 for trial in range(n_trials): model = train_func(train_data) score = evaluate_func(model) results['baseline'].append(score) # 测试每种增强方法 for aug_func in augmentation_funcs: aug_name = aug_func.__name__ results['augmented'][aug_name] = [] for trial in range(n_trials): # 对训练数据应用增强 augmented_data = [aug_func(x) for x in train_data] # 合并原始数据和增强数据 combined = train_data + augmented_data model = train_func(combined) score = evaluate_func(model) results['augmented'][aug_name].append(score) # 汇总结果 baseline_avg = np.mean(results['baseline']) print(f"基线（无增强）: {baseline_avg:.4f}") print(f"增强方法对比:") for aug_name, scores in results['augmented'].items(): avg = np.mean(scores) improvement = avg - baseline_avg print(f" {aug_name}: {avg:.4f} ({'+' if improvement >= 0 else ''}{improvement:.4f})") return results ``` ### 五、数据集划分 数据集划分看似简单，实际操作中有很多坑。 ```python from sklearn.model_selection import train_test_split, StratifiedKFold import hashlib class DatasetSplitter: """ 数据集划分工具 支持多种划分策略，确保数据泄露检测 """ def __init__(self, df: pd.DataFrame, label_col: str = None, id_col: str = None): self.df = df self.label_col = label_col self.id_col = id_col def stratified_split(self, train_ratio=0.7, val_ratio=0.15, test_ratio=0.15, random_state=42) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """ 分层划分：保证每个子集中各类别的比例一致 这是最常用的划分方法，特别适用于： - 类别不平衡的数据集 - 小数据集（随机划分可能导致某个子集缺少某个类别） """ assert abs(train_ratio + val_ratio + test_ratio - 1.0) < 1e-6 # 第一次划分：分出测试集 train_val, test = train_test_split( self.df, test_size=test_ratio, stratify=self.df[self.label_col] if self.label_col else None, random_state=random_state ) # 第二次划分：从训练+验证中分出验证集 relative_val_ratio = val_ratio / (train_ratio + val_ratio) train, val = train_test_split( train_val, test_size=relative_val_ratio, stratify=train_val[self.label_col] if self.label_col else None, random_state=random_state ) return train, val, test def group_aware_split(self, group_col: str, train_ratio=0.7, val_ratio=0.15, test_ratio=0.15, random_state=42) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """ 组感知划分：确保同一组的数据不会分散到不同子集 适用场景： - 医学影像（同一个病人的不同切片不能分散） - 用户行为（同一个用户的行为不能分散） - 视频帧（同一视频的不同帧不能分散） 如果不这样做，模型会"泄露"地学到同一组数据的共同特征， 导致验证集上的性能虚高，实际部署后性能骤降。 """ groups = self.df[group_col].unique() n_groups = len(groups) np.random.seed(random_state) np.random.shuffle(groups) n_train_groups = int(n_groups * train_ratio) n_val_groups = int(n_groups * val_ratio) train_groups = groups[:n_train_groups] val_groups = groups[n_train_groups:n_train_groups + n_val_groups] test_groups = groups[n_train_groups + n_val_groups:] train = self.df[self.df[group_col].isin(train_groups)] val = self.df[self.df[group_col].isin(val_groups)] test = self.df[self.df[group_col].isin(test_groups)] print(f"组感知划分结果:") print(f" 训练组数: {len(train_groups)}, 样本数: {len(train)}") print(f" 验证组数: {len(val_groups)}, 样本数: {len(val)}") print(f" 测试组数: {len(test_groups)}, 样本数: {len(test)}") return train, val, test def temporal_split(self, time_col: str, train_end: str, val_end: str) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """ 时间序列划分：按时间先后顺序划分 适用场景： - 时间序列预测 - 推荐系统（按时间划分才能模拟真实场景） - 任何有 temporal dependency 的数据 核心原则：绝不能用未来的数据训练来预测过去 """ self.df[time_col] = pd.to_datetime(self.df[time_col]) train_end = pd.to_datetime(train_end) val_end = pd.to_datetime(val_end) train = self.df[self.df[time_col] <= train_end] val = self.df[(self.df[time_col] > train_end) & (self.df[time_col] <= val_end)] test = self.df[self.df[time_col] > val_end] print(f"时间划分结果:") print(f" 训练集: {train[time_col].min()} ~ {train[time_col].max()} ({len(train)}条)") print(f" 验证集: {val[time_col].min()} ~ {val[time_col].max()} ({len(val)}条)") print(f" 测试集: {test[time_col].min()} ~ {test[time_col].max()} ({len(test)}条)") return train, val, test def detect_data_leakage(self, train: pd.DataFrame, test: pd.DataFrame, check_cols: List[str] = None) -> dict: """ 数据泄露检测 检查训练集和测试集之间是否存在不应该出现的重叠 """ results = {} # 检查行级重复 if check_cols: train_keys = set(train[check_cols].apply(tuple, axis=1)) test_keys = set(test[check_cols].apply(tuple, axis=1)) overlap = train_keys & test_keys results['row_overlap'] = { 'count': len(overlap), 'train_total': len(train), 'test_total': len(test), 'leakage_ratio': len(overlap) / len(test) if len(test) > 0 else 0, 'severity': 'high' if len(overlap) > len(test) * 0.01 else 'low' } # 检查ID列重叠 if self.id_col and self.id_col in train.columns: train_ids = set(train[self.id_col]) test_ids = set(test[self.id_col]) id_overlap = train_ids & test_ids results['id_overlap'] = { 'count': len(id_overlap), 'severity': 'high' if len(id_overlap) > 0 else 'none' } return results # 使用示例 np.random.seed(42) n = 5000 # 模拟一个医疗影像数据集（同一病人有多张影像） patient_ids = [f'P{i:04d}' for i in range(500)] data = [] for pid in patient_ids: n_scans = np.random.randint(2, 15) for j in range(n_scans): data.append({ 'patient_id': pid, 'scan_id': f'{pid}_S{j:03d}', 'label': np.random.choice(['benign', 'malignant'], p=[0.7, 0.3]), 'scan_date': pd.Timestamp('2023-01-01') + pd.Timedelta(days=np.random.randint(0, 730)) }) df_medical = pd.DataFrame(data) print(f"医疗影像数据集: {len(df_medical)} 条记录, {df_medical['patient_id'].nunique()} 个病人") splitter = DatasetSplitter(df_medical, label_col='label', id_col='scan_id') # 错误做法：随机划分（会导致数据泄露） print("\n=== 随机划分（有数据泄露风险）===") train_wrong, val_wrong, test_wrong = splitter.stratified_split() leakage = splitter.detect_data_leakage(train_wrong, test_wrong, check_cols=['patient_id']) print(f"病人级别泄露: {leakage}") # 正确做法：组感知划分 print("\n=== 组感知划分（无数据泄露）===") train_right, val_right, test_right = splitter.group_aware_split(group_col='patient_id') leakage = splitter.detect_data_leakage(train_right, test_right, check_cols=['patient_id']) print(f"病人级别泄露: {leakage}") ``` ## 代码示例（完整可运行的 Python） 下面是一个完整的数据工程流水线，将上述所有步骤串联起来： ```python """ 完整的数据工程流水线示例 从原始数据到可用于训练的干净数据集 """ import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional from dataclasses import dataclass, field from datetime import datetime import json import hashlib # ============================================================ # 配置 # ============================================================ @dataclass class DataPipelineConfig: """数据管线配置""" name: str = "default_pipeline" version: str = "1.0.0" # 数据质量阈值 missing_threshold: float = 0.3 # 缺失率超过此值的列删除 outlier_method: str = "iqr" # 'iqr' or 'zscore' outlier_factor: float = 1.5 # 标注质量控制 min_annotator_agreement: float = 0.7 # Cohen's Kappa最低要求 # 数据增强 augmentation_enabled: bool = True augmentation_ratio: float = 0.5 # 增强数据与原始数据的比例 augmentation_methods: List[str] = field(default_factory=lambda: [ 'synonym_replacement', 'random_deletion', 'random_swap' ]) # 数据集划分 train_ratio: float = 0.7 val_ratio: float = 0.15 test_ratio: float = 0.15 stratify: bool = True # 元信息 created_at: str = field(default_factory=lambda: datetime.now().isoformat()) description: str = "" # ============================================================ # 数据管线 # ============================================================ class DataPipeline: """ 端到端的数据工程管线 流程： 原始数据 → 质量评估 → 清洗 → 标注质量控制 → 增强 → 划分 → 输出 每一步都记录详细的日志和统计信息 """ def __init__(self, config: DataPipelineConfig): self.config = config self.log = [] self.stats = {} def _log(self, step: str, message: str, data: dict = None): entry = { 'timestamp': datetime.now().isoformat(), 'step': step, 'message': message, 'data': data or {} } self.log.append(entry) print(f"[{step}] {message}") def run(self, raw_data: pd.DataFrame, label_col: str = None, group_col: str = None) -> dict: """ 运行完整管线 返回包含训练集、验证集、测试集和完整报告的字典 """ self._log('START', f'管线启动: {self.config.name} v{self.config.version}') self._log('START', f'输入数据: {raw_data.shape}') # Step 1: 初始质量评估 quality_report = self._assess_quality(raw_data) self.stats['initial_quality'] = quality_report # Step 2: 数据清洗 cleaned = self._clean(raw_data) self.stats['cleaning'] = { 'input_rows': len(raw_data), 'output_rows': len(cleaned), 'rows_removed': len(raw_data) - len(cleaned) } # Step 3: 数据增强 if self.config.augmentation_enabled: augmented = self._augment(cleaned, label_col) self.stats['augmentation'] = { 'original_count': len(cleaned), 'augmented_count': len(augmented), 'total_count': len(cleaned) + len(augmented) } final_data = pd.concat([cleaned, augmented], ignore_index=True) else: final_data = cleaned # Step 4: 数据集划分 splits = self._split(final_data, label_col, group_col) self.stats['split'] = { 'train_size': len(splits['train']), 'val_size': len(splits['val']), 'test_size': len(splits['test']), } # Step 5: 最终质量评估 final_quality = self._assess_quality(final_data) self.stats['final_quality'] = final_quality # Step 6: 数据泄露检测 if group_col: leakage = self._check_leakage(splits['train'], splits['test'], group_col) self.stats['leakage_check'] = leakage self._log('DONE', f'管线完成. 输出: train={len(splits["train"])}, val={len(splits["val"])}, test={len(splits["test"])}') return { 'train': splits['train'], 'val': splits['val'], 'test': splits['test'], 'report': self._generate_report() } def _assess_quality(self, df: pd.DataFrame) -> dict: """数据质量评估""" self._log('QUALITY', f'评估数据质量...') missing_pct = df.isna().mean().to_dict() n_duplicates = df.duplicated().sum() report = { 'shape': df.shape, 'missing_percentage': {k: round(v, 4) for k, v in missing_pct.items()}, 'duplicate_rows': int(n_duplicates), 'duplicate_rate': round(n_duplicates / len(df), 4), 'column_types': df.dtypes.astype(str).to_dict() } self._log('QUALITY', f'缺失率: {max(missing_pct.values()):.2%}, 重复率: {report["duplicate_rate"]:.2%}') return report def _clean(self, df: pd.DataFrame) -> pd.DataFrame: """数据清洗""" self._log('CLEAN', f'开始清洗 {len(df)} 条数据...') result = df.copy() # 删除高缺失列 missing_pct = result.isna().mean() high_missing_cols = missing_pct[missing_pct > self.config.missing_threshold].index.tolist() if high_missing_cols: result = result.drop(columns=high_missing_cols) self._log('CLEAN', f'删除高缺失列: {high_missing_cols}') # 去重 before = len(result) result = result.drop_duplicates(keep='first') self._log('CLEAN', f'去重: {before} → {len(result)}') # 删除包含缺失值的行（简化策略） before = len(result) result = result.dropna() self._log('CLEAN', f'删除缺失行: {before} → {len(result)}') return result def _augment(self, df: pd.DataFrame, label_col: str = None) -> pd.DataFrame: """数据增强""" self._log('AUGMENT', f'生成增强数据 (比例: {self.config.augmentation_ratio})') n_augment = int(len(df) * self.config.augmentation_ratio) augment_indices = np.random.choice(len(df), size=n_augment, replace=True) augmented_rows = [] for idx in augment_indices: row = df.iloc[idx].copy() # 添加噪声作为简化的增强 numeric_cols = row.select_dtypes(include=[np.number]).index for col in numeric_cols: noise = np.random.normal(0, 0.01 * abs(row[col]) + 0.001) row[col] = row[col] + noise row.name = f"aug_{idx}" augmented_rows.append(row) if augmented_rows: return pd.DataFrame(augmented_rows) return pd.DataFrame() def _split(self, df: pd.DataFrame, label_col: str = None, group_col: str = None) -> dict: """数据集划分""" self._log('SPLIT', f'划分数据集 ({self.config.train_ratio}/{self.config.val_ratio}/{self.config.test_ratio})') if group_col and group_col in df.columns: # 组感知划分 groups = df[group_col].unique() np.random.seed(42) np.random.shuffle(groups) n_train = int(len(groups) * self.config.train_ratio) n_val = int(len(groups) * self.config.val_ratio) train_groups = groups[:n_train] val_groups = groups[n_train:n_train + n_val] test_groups = groups[n_train + n_val:] train = df[df[group_col].isin(train_groups)] val = df[df[group_col].isin(val_groups)] test = df[df[group_col].isin(test_groups)] else: # 标准划分 stratify = df[label_col] if (label_col and self.config.stratify) else None test_size = self.config.test_ratio train_val, test = train_test_split( df, test_size=test_size, stratify=stratify, random_state=42 ) val_ratio_adjusted = self.config.val_ratio / (self.config.train_ratio + self.config.val_ratio) stratify_tv = train_val[label_col] if (label_col and self.config.stratify) else None train, val = train_test_split( train_val, test_size=val_ratio_adjusted, stratify=stratify_tv, random_state=42 ) return {'train': train, 'val': val, 'test': test} def _check_leakage(self, train: pd.DataFrame, test: pd.DataFrame, group_col: str) -> dict: """数据泄露检测""" train_groups = set(train[group_col].unique()) test_groups = set(test[group_col].unique()) overlap = train_groups & test_groups result = { 'train_groups': len(train_groups), 'test_groups': len(test_groups), 'overlapping_groups': len(overlap), 'leakage_detected': len(overlap) > 0 } if overlap: self._log('LEAKAGE', f'警告: 检测到 {len(overlap)} 个重叠的组!', {'severity': 'HIGH'}) else: self._log('LEAKAGE', '无数据泄露') return result def _generate_report(self) -> dict: """生成完整报告""" return { 'config': { 'name': self.config.name, 'version': self.config.version, 'created_at': self.config.created_at, }, 'stats': self.stats, 'log': self.log } # ============================================================ # 端到端运行示例 # ============================================================ # 1. 创建模拟数据 np.random.seed(42) n = 10000 patient_ids = [f'P{i:04d}' for i in range(1000)] data = [] for pid in patient_ids: n_records = np.random.randint(1, 20) for j in range(n_records): data.append({ 'patient_id': pid, 'feature_1': np.random.normal(50, 10), 'feature_2': np.random.normal(100, 20), 'feature_3': np.random.choice(['A', 'B', 'C'], p=[0.5, 0.3, 0.2]), 'label': np.random.choice([0, 1], p=[0.7, 0.3]), }) # 注入一些质量问题 df = pd.DataFrame(data) # 随机插入缺失值 for col in ['feature_1', 'feature_2']: mask = np.random.random(len(df)) < 0.03 df.loc[mask, col] = np.nan # 随机插入重复行 duplicates = df.sample(200) df = pd.concat([df, duplicates], ignore_index=True) print(f"原始数据: {df.shape}") print(f"缺失率:\n{df.isna().mean()}") print(f"重复行: {df.duplicated().sum()}") # 2. 配置并运行管线 config = DataPipelineConfig( name="medical_classification_v1", version="1.0.0", missing_threshold=0.5, augmentation_enabled=True, augmentation_ratio=0.3, train_ratio=0.7, val_ratio=0.15, test_ratio=0.15, stratify=True, ) pipeline = DataPipeline(config) result = pipeline.run(df, label_col='label', group_col='patient_id') # 3. 输出结果 print(f"\n最终数据集:") print(f" 训练集: {result['train'].shape}") print(f" 验证集: {result['val'].shape}") print(f" 测试集: {result['test'].shape}") print(f"\n标签分布:") for split_name in ['train', 'val', 'test']: dist = result[split_name]['label'].value_counts(normalize=True) print(f" {split_name}: {dist.to_dict()}") # 4. 输出管线报告 report = result['report'] print(f"\n管线统计:") print(f" 初始数据行数: {report['stats']['initial_quality']['shape'][0]}") print(f" 清洗后行数: {report['stats']['cleaning']['output_rows']}") print(f" 增强后总行数: {report['stats']['augmentation']['total_count']}") print(f" 数据泄露: {'是' if report['stats'].get('leakage_check', {}).get('leakage_detected') else '否'}") ``` ## 真实案例 ### 案例1：GPT-4的数据工程 OpenAI在GPT-4的技术报告中没有透露太多训练数据的细节，但从其前代模型的论文和公开信息中，可以拼凑出一些关键的数据工程决策： **数据收集**：GPT-4的训练数据来自多种来源的混合，包括网络爬取数据、书籍、学术论文、代码仓库等。关键是不同来源的数据质量差异巨大，需要仔细的过滤和混合比例调优。 **数据清洗**： - 去重：使用MinHash等算法在大规模数据集上去重，减少记忆效应 - 质量过滤：训练一个分类器来区分"高质量"文本和"低质量"文本 - 有害内容过滤：移除包含仇恨言论、暴力等有害内容的文本 - 个人信息脱敏：移除或脱敏个人身份信息 **数据混合比例**：不同领域的数据比例对模型能力有显著影响。增加代码数据的比例会提升模型的推理能力（即使是在非代码任务上）。这也是为什么后来的开源模型通常会在训练数据中加入大量的代码数据。 ### 案例2：Waymo的自动驾驶数据 Waymo的自动驾驶系统每天产生约1TB的传感器数据。这些数据需要经过极其严格的数据工程： **数据收集策略**： - 长尾场景挖掘：主动搜索和收集模型表现差的场景（雨天、夜间、施工区域） - 地理多样性：在多个城市收集数据，确保覆盖不同的道路布局和交通习惯 - 时间多样性：覆盖不同季节、不同时间段的数据 **标注流程**： - 自动标注（模型辅助）+ 人工标注（关键场景） - 3D边界框标注、语义分割、关键点标注 - 每帧标注时间：人工约15-30分钟，模型辅助约1-3分钟 - 多人标注 + 交叉验证，一致性要求 > 95% **质量控制**： - 每周标注质量审计 - 标注者间一致性追踪 - 自动化异常检测 ### 案例3：Tesla的自动标注 Tesla在2023年的AI Day上展示了其自动标注系统的进化： - **第一代**：人工标注2D边界框，每帧标注需要几分钟 - **第二代**：3D标注，在多个摄像头视角之间联合标注，效率提升3x - **第三代**：自动标注系统，利用已经训练好的模型在大规模未标注数据上生成伪标签，然后由人工审核。标注效率提升约1000x。 这个案例说明了数据工程的一个重要趋势：**模型本身正在成为数据工程的核心工具**。用模型辅助标注、用模型发现数据问题、用模型生成合成数据。 ## 权衡取舍以及何时不该使用 ### 数据清洗的权衡 **过度清洗的风险**：如果你把所有"异常值"都删除了，可能会删除掉真正的少数群体数据。在欺诈检测中，欺诈行为本身就是异常值——如果你把它们清洗掉了，模型就学不到任何东西了。 **权衡**：数据清洗应该基于业务理解和统计证据的结合。不要盲目地用IQR方法删除所有异常值，要先问"这个异常值是数据错误，还是反映了真实的但罕见的情况？" ### 数据增强的权衡 **增强语义漂移**：当增强操作改变了数据的语义时，模型会学到错误的知识。在情感分析中，把"差"替换成"好"看起来是同义词替换，但实际上改变了情感极性。 **权衡**： - 图像增强的语义漂移风险较低（翻转、旋转不太可能改变语义） - 文本增强的语义漂移风险较高（任何操作都可能改变语义） - NLP任务中，回译（back-translation）是最安全的增强方法 - 混合类增强（Mixup, CutMix）只适用于分类任务，不适用于检测/分割 ### 合成数据的权衡 **分布偏差**：合成数据的分布可能无法完全匹配真实数据的分布。用GPT-4生成的对话数据，在表达方式上可能与真实用户的表达方式有显著差异。 **模式崩溃**：如果用模型生成数据再用另一个模型训练，两个模型可能共享相同的偏见和缺陷，形成正反馈循环。 **权衡**： - 合成数据最适合用于补充稀缺场景的数据 - 合成数据不应该完全替代真实数据 - 始终在真实数据上做最终验证 - 对合成数据的质量要进行严格的自动评估 ### 何时不该花太多时间在数据工程上 数据工程是重要的，但也有一些场景下，你应该快速迭代而不是追求完美的数据： 1. **探索阶段**：当你还不确定问题定义是否正确时，用简单的数据快速跑一个baseline比花三周做数据清洗更有价值。 2. **原型验证**：在验证一个idea是否可行时，用公开数据集快速验证比从头搭建数据管线更高效。 3. **模型架构探索**：当你不确定哪种模型架构更适合你的任务时，先用标准数据集找到合适的架构，再在自有数据上精调。 ## 关键要点 1. **数据质量决定模型上限**。一条满是噪声的训练数据，就像一本满是错误的教科书。不管学生（模型）有多聪明，学出来的知识都是有问题的。Andrew Ng的Data-Centric AI范式强调：固定模型架构，集中精力提高数据质量，ROI远大于不断换模型。 2. **数据清洗不是一次性工作**。数据质量会随着时间退化（数据漂移），新的数据源会引入新的质量问题，模型的错误模式会变化。你需要建立一个持续的数据质量监控和清洗流程。 3. **数据标注是人工成本最高的环节，也是最容易出错的环节**。投资于标注指南的编写、标注者的培训、标注质量的度量（Cohen's Kappa, Fleiss' Kappa），可以在后续节省大量重新训练的成本。 4. **数据增强是最廉价也最有效的正则化手段**。但增强方法的选择要与任务匹配：图像可以用CutMix/Mixup，NLP用回译/同义词替换，混合类增强不适用于所有任务。 5. **数据泄露是静默的杀手**。如果你的训练集和测试集有重叠（直接的行重叠，或更隐蔽的组级别重叠），验证集上的性能会虚高，部署后性能骤降。组感知划分和时间划分是防范数据泄露的关键方法。 6. **数据工程的工具正在被AI本身重塑**。模型辅助标注、自动数据清洗、合成数据生成——这些趋势意味着数据工程师需要同时理解数据质量和模型训练。 7. **数据集划分不是随机切一刀**。分层划分保证类别比例一致，组感知划分防止数据泄露，时间划分模拟真实部署场景。选择错误的划分方法，会导致模型评估完全不可信。 ## 延伸阅读 **论文**： - "Data-centric AI: Perspectives and Challenges" (Borges et al., 2025) — 以数据为中心的AI框架 - "The unreasonable effectiveness of data" (Halevy et al., 2009) — 数据比算法更重要的经典论证 - "Chinchilla: Training Language Models to Compute and Communicate" (DeepMind, 2022) — 证明数据量的重要性 - "LIMA: Less Is More for Alignment" (Zhou et al., 2023) — 1000条高质量数据的力量 - "mixup: Beyond Empirical Risk Minimization" (Zhang et al., 2018) — Mixup增强方法 - "CutMix: Regularization Strategy" (Yun et al., 2019) — CutMix增强方法 - "A survey on Image Data Augmentation for Deep Learning" (Shorten & Khoshgoftaar, 2019) — 图像增强综述 **工具**： - Cleanlab — 自动发现数据标注错误 - Label Studio — 开源数据标注平台 - Great Expectations — 数据质量验证框架 - DVC (Data Version Control) — 数据版本管理 -Weights & Biases (W&B) — 实验跟踪与数据集版本管理 **教程与指南**： - Andrew Ng "Data-Centric AI" 课程 (DeepLearning.AI) - Google "Rules of Machine Learning" — ML工程最佳实践 - Scale AI "Data Labeling Guide" — 数据标注行业指南