2.14 数据工程

一句话总结：Garbage in, garbage out——数据质量决定 AI 应用的上限。

📊 学习进度
状态：⬜ 未开始
预计时长：4-6 小时
已完成：0/4 个模块
在整体流程中的位置：AI 应用开发·第 2 阶段

📍 本章定位
服务方案：方案 1（核心 80%）/ 方案 2（重要 60%）
学习方式：⭐ 必学
在流程中的作用：数据是 AI 的燃料，数据工程通常占 60-80% 的工作量
核心知识点：数据采集、清洗、标注、版本管理
预计时长：4-6 小时
完成后能做什么：能为 AI 应用准备高质量的数据集

什么是数据工程？

核心定义

数据工程是将原始数据转化为可用于 AI 模型训练的高质量数据集的系统性工程。它不是简单的"数据处理"，而是包含数据采集、清洗、标注、版本管理的完整流程。

为什么占 60-80% 工作量？

阶段	占比	原因
数据采集	15-20%	寻找、获取、整合数据源
数据清洗	25-30%	去重、去噪、格式统一、异常处理
数据标注	15-25%	人工或半自动标注，质量控制
数据版本管理	5-10%	版本控制、变更追踪、回滚机制

关键认知：AI 模型的上限由数据质量决定，而非模型复杂度。好的数据 + 简单模型 > 差数据 + 复杂模型。

人机分工

环节	谁做	重要度	说明
数据标准定义	🧑 人	⭐⭐⭐⭐⭐	业务理解决定数据质量
数据采集策略	🧑 人	⭐⭐⭐⭐	决定数据来源和范围
数据清洗执行	🤖 AI	⭐⭐⭐⭐	人定规则，AI 执行
数据标注	🤖 AI + 🧑 审核	⭐⭐⭐⭐	AI 初标，人做质检
数据版本管理	🤖 AI	⭐⭐⭐	工具自动管理

数据工程四步流程

Step 1：数据采集

人机分工：🧑 人决定采集什么 / 🤖 AI 执行采集

数据来源

来源	适用场景	成本	质量
公开数据集	快速验证、通用场景	免费	中
API 抓取	特定领域数据	低	中高
业务系统日志	已有业务场景	免费	高
人工标注	高质量定制数据	高	最高

API 抓取示例

python

import requests
import pandas as pd
from typing import List, Dict

def fetch_data_from_api(api_url: str, headers: Dict = None) -> pd.DataFrame:
    """
    从 API 获取数据并转换为 DataFrame
    
    Args:
        api_url: API 端点 URL
        headers: 请求头（包含认证信息）
    
    Returns:
        pd.DataFrame: 包含 API 数据的 DataFrame
    """
    try:
        response = requests.get(api_url, headers=headers, timeout=30)
        response.raise_for_status()  # 检查请求是否成功
        
        data = response.json()
        
        # 转换为 DataFrame
        if isinstance(data, list):
            df = pd.DataFrame(data)
        elif isinstance(data, dict) and 'results' in data:
            df = pd.DataFrame(data['results'])
        else:
            df = pd.DataFrame([data])
        
        print(f"成功获取 {len(df)} 条数据")
        return df
    
    except requests.exceptions.RequestException as e:
        print(f"API 请求失败: {e}")
        return pd.DataFrame()

# 使用示例：获取加密货币市场数据
api_url = "https://api.coingecko.com/api/v3/coins/markets"
params = {
    "vs_currency": "usd",
    "order": "market_cap_desc",
    "per_page": 100,
    "page": 1
}

df = fetch_data_from_api(api_url, params)
print(df.head())

网页爬取示例

python

import requests
from bs4 import BeautifulSoup
import pandas as pd
import time

def scrape_web_data(url: str, selectors: Dict) -> pd.DataFrame:
    """
    网页爬取数据
    
    Args:
        url: 目标网页 URL
        selectors: CSS 选择器字典 {"字段名": "选择器"}
    
    Returns:
        pd.DataFrame: 爬取的数据
    """
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=30)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.content, 'html.parser')
        
        data = {}
        for field_name, selector in selectors.items():
            elements = soup.select(selector)
            data[field_name] = [elem.get_text(strip=True) for elem in elements]
        
        # 确保所有字段长度一致
        min_len = min(len(v) for v in data.values())
        data = {k: v[:min_len] for k, v in data.items()}
        
        df = pd.DataFrame(data)
        print(f"成功爬取 {len(df)} 条数据")
        return df
    
    except Exception as e:
        print(f"爬取失败: {e}")
        return pd.DataFrame()

# 使用示例
url = "https://example.com/data"
selectors = {
    "title": "h2.title",
    "price": "span.price",
    "description": "p.description"
}

df = scrape_web_data(url, selectors)

Step 2：数据清洗

人机分工：🧑 人定清洗规则 / 🤖 AI 执行清洗

核心任务

任务	工具	说明
去重	pandas / dedupe	移除重复记录
去噪	正则 + AI 过滤	移除无关内容
格式统一	pandas	日期、编码、单位统一
异常值检测	统计方法 + AI	识别和处理异常

完整清洗流程

数据清洗代码示例

python

import pandas as pd
import numpy as np
from typing import Tuple

class DataCleaner:
    """数据清洗工具类"""
    
    def __init__(self, df: pd.DataFrame):
        self.df = df.copy()
        self.cleaning_log = []
    
    def overview(self) -> Dict:
        """数据概览"""
        info = {
            "总行数": len(self.df),
            "总列数": len(self.df.columns),
            "缺失值": self.df.isnull().sum().sum(),
            "重复行": self.df.duplicated().sum(),
            "数据类型": self.df.dtypes.to_dict()
        }
        return info
    
    def handle_missing_values(self, strategy: str = "auto") -> 'DataCleaner':
        """
        处理缺失值
        
        Args:
            strategy: 处理策略
                - "auto": 自动选择（数值用中位数，类别用众数）
                - "drop": 删除含缺失值的行
                - "fill_mean": 用均值填充
                - "fill_median": 用中位数填充
        """
        initial_missing = self.df.isnull().sum().sum()
        
        if strategy == "auto":
            for col in self.df.columns:
                if self.df[col].dtype in ['int64', 'float64']:
                    self.df[col].fillna(self.df[col].median(), inplace=True)
                else:
                    self.df[col].fillna(self.df[col].mode()[0], inplace=True)
        
        elif strategy == "drop":
            self.df.dropna(inplace=True)
        
        elif strategy == "fill_mean":
            numeric_cols = self.df.select_dtypes(include=[np.number]).columns
            self.df[numeric_cols] = self.df[numeric_cols].fillna(self.df[numeric_cols].mean())
        
        elif strategy == "fill_median":
            numeric_cols = self.df.select_dtypes(include=[np.number]).columns
            self.df[numeric_cols] = self.df[numeric_cols].fillna(self.df[numeric_cols].median())
        
        final_missing = self.df.isnull().sum().sum()
        self.cleaning_log.append(f"缺失值处理: {initial_missing} -> {final_missing}")
        
        return self
    
    def remove_duplicates(self, subset: List[str] = None) -> 'DataCleaner':
        """去除重复值"""
        initial_rows = len(self.df)
        self.df.drop_duplicates(subset=subset, inplace=True)
        removed = initial_rows - len(self.df)
        self.cleaning_log.append(f"去除重复: 移除 {removed} 行")
        return self
    
    def handle_outliers(self, columns: List[str], method: str = "iqr") -> 'DataCleaner':
        """
        处理异常值
        
        Args:
            columns: 要检查的列
            method: 检测方法
                - "iqr": 四分位距法
                - "zscore": Z分数法
        """
        for col in columns:
            if method == "iqr":
                Q1 = self.df[col].quantile(0.25)
                Q3 = self.df[col].quantile(0.75)
                IQR = Q3 - Q1
                lower = Q1 - 1.5 * IQR
                upper = Q3 + 1.5 * IQR
                
                outliers = ((self.df[col] < lower) | (self.df[col] > upper)).sum()
                self.df = self.df[(self.df[col] >= lower) & (self.df[col] <= upper)]
                
            elif method == "zscore":
                from scipy import stats
                z_scores = np.abs(stats.zscore(self.df[col]))
                outliers = (z_scores > 3).sum()
                self.df = self.df[z_scores <= 3]
            
            self.cleaning_log.append(f"异常值处理 ({col}): 移除 {outliers} 个异常值")
        
        return self
    
    def standardize_formats(self, date_columns: List[str] = None) -> 'DataCleaner':
        """统一数据格式"""
        if date_columns:
            for col in date_columns:
                self.df[col] = pd.to_datetime(self.df[col], errors='coerce')
                self.cleaning_log.append(f"日期格式统一: {col}")
        
        # 字符串列去除首尾空格
        string_cols = self.df.select_dtypes(include=['object']).columns
        for col in string_cols:
            self.df[col] = self.df[col].str.strip()
        
        return self
    
    def get_cleaned_data(self) -> Tuple[pd.DataFrame, List[str]]:
        """获取清洗后的数据和日志"""
        return self.df, self.cleaning_log

# 使用示例
df = pd.read_csv("raw_data.csv")

cleaner = DataCleaner(df)
cleaned_df, log = (
    cleaner
    .handle_missing_values(strategy="auto")
    .remove_duplicates()
    .handle_outliers(columns=["price", "volume"])
    .standardize_formats(date_columns=["created_at", "updated_at"])
    .get_cleaned_data()
)

print("清洗日志:")
for entry in log:
    print(f"  - {entry}")

关键认知：花在数据上的时间永远不会浪费。好的数据 + 简单模型 > 差数据 + 复杂模型。

Step 3：数据标注

人机分工：🤖 AI 初步标注 / 🧑 人做质检

标注流程

标注策略

策略	适用场景	成本	说明
自己标注	小规模、高要求	时间成本高	完全控制质量
外包标注	大规模、标准化	资金成本高	需要详细规范
主动学习	持续优化	最优 ROI	AI 辅助选择样本
半监督学习	标注数据少	中等	利用未标注数据

标注工具使用示例

Label Studio 安装与配置

bash

# 安装 Label Studio
pip install label-studio

# 启动服务
label-studio start

# 访问 http://localhost:8080

Label Studio Python API

python

from label_studio_sdk import Client

# 连接 Label Studio
ls = Client(url='http://localhost:8080', API_KEY='your-api-key')

# 创建项目
project = ls.start_project(
    title='文本情感分析',
    label_config='''
    <View>
        <Labels name="sentiment" toName="text">
            <Label value="正面" background="#00ff00"/>
            <Label value="负面" background="#ff0000"/>
            <Label value="中性" background="#ffff00"/>
        </Labels>
        <Text name="text" value="$text"/>
    </View>
    '''
)

# 导入数据
project.import_tasks([
    {"text": "这个产品非常好用！"},
    {"text": "服务太差了，再也不买了。"},
    {"text": "还行吧，一般般。"}
])

# 获取标注结果
results = project.get_labeled_tasks()

主动学习示例

python

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import uncertainty_measure

class ActiveLearner:
    """主动学习：选择最有价值的样本进行标注"""
    
    def __init__(self, model, X_labeled, y_labeled, X_unlabeled):
        self.model = model
        self.X_labeled = X_labeled
        self.y_labeled = y_labeled
        self.X_unlabeled = X_unlabeled
    
    def select_samples(self, n_samples: int = 10) -> np.ndarray:
        """
        选择最不确定的样本进行标注
        
        Args:
            n_samples: 选择样本数量
        
        Returns:
            选中样本的索引
        """
        # 训练模型
        self.model.fit(self.X_labeled, self.y_labeled)
        
        # 预测概率
        probas = self.model.predict_proba(self.X_unlabeled)
        
        # 计算不确定性（熵）
        uncertainty = -np.sum(probas * np.log(probas + 1e-10), axis=1)
        
        # 选择最不确定的样本
        selected_indices = np.argsort(uncertainty)[-n_samples:]
        
        return selected_indices
    
    def update(self, X_new, y_new):
        """更新标注数据"""
        self.X_labeled = np.vstack([self.X_labeled, X_new])
        self.y_labeled = np.concatenate([self.y_labeled, y_new])

# 使用示例
learner = ActiveLearner(
    model=RandomForestClassifier(n_estimators=100),
    X_labeled=labeled_data,
    y_labeled=labeled_labels,
    X_unlabeled=unlabeled_data
)

# 选择最有价值的样本
indices = learner.select_samples(n_samples=20)
samples_to_label = unlabeled_data[indices]

Step 4：数据版本管理

人机分工：🤖 AI 自动管理

为什么需要数据版本管理？

问题	没有版本管理	有版本管理
数据回滚	手动查找备份	一键回滚到任意版本
实验复现	无法确定使用了哪版数据	精确记录数据版本
团队协作	数据冲突、版本混乱	统一版本控制
审计追踪	无法追溯数据变更	完整变更历史

DVC 使用示例

安装与初始化

bash

# 安装 DVC
pip install dvc

# 初始化 DVC
dvc init

# 配置远程存储（以 S3 为例）
dvc remote add -d myremote s3://mybucket/dvcstore
dvc remote modify myremote access_key_id your_access_key
dvc remote modify myremote secret_access_key your_secret_key

数据版本控制

bash

# 添加数据文件到 DVC 追踪
dvc add data/train.csv
dvc add data/test.csv

# 提交 DVC 文件到 Git
git add data/train.csv.dvc data/test.csv.dvc .gitignore
git commit -m "添加训练和测试数据"

# 推送数据到远程存储
dvc push

# 切换到某个版本
git checkout v1.0
dvc checkout

# 拉取特定版本的数据
git checkout v2.0
dvc pull

DVC Pipeline

yaml

# dvc.yaml - 定义数据处理流水线
stages:
  prepare:
    cmd: python prepare.py
    deps:
      - data/raw
      - prepare.py
    outs:
      - data/prepared
  
  train:
    cmd: python train.py
    deps:
      - data/prepared
      - train.py
    outs:
      - models/model.pkl
    metrics:
      - metrics.json

bash

# 运行整个流水线
dvc repro

# 查看指标对比
dvc metrics show
dvc metrics diff

数据质量评估

评估指标

指标	定义	计算方法
完整性	数据无缺失	非空值数 / 总值数
准确性	数据正确无误	正确值数 / 总值数
一致性	数据在各处一致	一致记录数 / 总记录数
时效性	数据是否过期	最新记录占比
唯一性	无重复记录	唯一记录数 / 总记录数

数据质量检查流程

质量评估代码

python

import pandas as pd
from typing import Dict

def assess_data_quality(df: pd.DataFrame) -> Dict:
    """
    评估数据质量
    
    Args:
        df: 待评估的 DataFrame
    
    Returns:
        质量评估报告
    """
    report = {
        "completeness": {},
        "uniqueness": {},
        "consistency": {},
        "overall_score": 0
    }
    
    # 完整性评估
    for col in df.columns:
        non_null = df[col].notna().sum()
        report["completeness"][col] = {
            "non_null_count": non_null,
            "null_count": len(df) - non_null,
            "completeness_rate": non_null / len(df)
        }
    
    # 唯一性评估
    for col in df.columns:
        unique_count = df[col].nunique()
        report["uniqueness"][col] = {
            "unique_count": unique_count,
            "duplicate_count": len(df) - unique_count,
            "uniqueness_rate": unique_count / len(df)
        }
    
    # 一致性检查（示例：日期格式）
    date_cols = df.select_dtypes(include=['datetime64']).columns
    for col in date_cols:
        try:
            pd.to_datetime(df[col])
            report["consistency"][col] = {"status": "consistent"}
        except:
            report["consistency"][col] = {"status": "inconsistent"}
    
    # 计算总体质量分数
    completeness_scores = [v["completeness_rate"] for v in report["completeness"].values()]
    report["overall_score"] = sum(completeness_scores) / len(completeness_scores)
    
    return report

# 使用示例
df = pd.read_csv("data.csv")
quality_report = assess_data_quality(df)

print(f"数据质量分数: {quality_report['overall_score']:.2%}")
print("\n各列完整性:")
for col, info in quality_report["completeness"].items():
    print(f"  {col}: {info['completeness_rate']:.2%}")

数据集划分策略

划分比例

数据集	比例	用途
训练集	70-80%	模型学习
验证集	10-15%	超参数调优、模型选择
测试集	10-15%	最终评估、性能测试

划分代码示例

python

import pandas as pd
from sklearn.model_selection import train_test_split
from typing import Tuple

def split_dataset(
    df: pd.DataFrame,
    target_col: str,
    test_size: float = 0.2,
    val_size: float = 0.1,
    random_state: int = 42,
    stratify: bool = True
) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """
    划分数据集为训练集、验证集、测试集
    
    Args:
        df: 原始数据集
        target_col: 目标列名
        test_size: 测试集比例
        val_size: 验证集比例
        random_state: 随机种子
        stratify: 是否分层抽样
    
    Returns:
        (train_df, val_df, test_df)
    """
    stratify_col = df[target_col] if stratify else None
    
    # 先划分出测试集
    train_val_df, test_df = train_test_split(
        df,
        test_size=test_size,
        random_state=random_state,
        stratify=stratify_col
    )
    
    # 再从训练验证集中划分出验证集
    if stratify:
        stratify_col = train_val_df[target_col]
    
    train_df, val_df = train_test_split(
        train_val_df,
        test_size=val_size / (1 - test_size),
        random_state=random_state,
        stratify=stratify_col
    )
    
    print(f"训练集: {len(train_df)} 条 ({len(train_df)/len(df):.1%})")
    print(f"验证集: {len(val_df)} 条 ({len(val_df)/len(df):.1%})")
    print(f"测试集: {len(test_df)} 条 ({len(test_df)/len(df):.1%})")
    
    return train_df, val_df, test_df

# 使用示例
df = pd.read_csv("data.csv")
train_df, val_df, test_df = split_dataset(df, target_col="label")

# 保存划分后的数据集
train_df.to_csv("data/train.csv", index=False)
val_df.to_csv("data/val.csv", index=False)
test_df.to_csv("data/test.csv", index=False)

注意事项

分层抽样：分类任务中，确保各数据集类别分布一致
时间序列：按时间顺序划分，不能随机打乱
数据泄露：测试集不能参与任何训练或预处理过程

数据增强

为什么需要数据增强？

问题	数据增强的效果	适用场景
数据量不足	通过变换增加有效样本数	小样本学习
类别不平衡	对少数类进行增强	分类任务
过拟合	增加数据多样性，提升泛化能力	所有场景
领域适配	生成领域特定数据	垂直场景

数据增强策略选择流程

文本数据增强

python

import nlpaug.augmenter.word as naw
import nlpaug.augmenter.sentence as nas

class TextAugmentor:
    """文本数据增强工具"""
    
    def __init__(self):
        # 同义词替换
        self.synonym_aug = naw.SynonymAug(aug_src='wordnet')
        # 随机插入
        self.insert_aug = naw.RandomWordAug(action="insert")
        # 随机删除
        self.delete_aug = naw.RandomWordAug(action="delete")
        # 回译
        self.back_translation_aug = naw.BackTranslationAug(
            from_model_name='facebook/wmt19-en-de',
            to_model_name='facebook/wmt19-de-en'
        )
    
    def augment(self, text: str, n: int = 3) -> List[str]:
        """
        对文本进行多种增强
        
        Args:
            text: 原始文本
            n: 每种方法生成的样本数
        
        Returns:
            增强后的文本列表
        """
        augmented = []
        
        # 同义词替换
        for _ in range(n):
            augmented.append(self.synonym_aug.augment(text))
        
        # 随机插入
        for _ in range(n):
            augmented.append(self.insert_aug.augment(text))
        
        # 随机删除
        for _ in range(n):
            augmented.append(self.delete_aug.augment(text))
        
        return augmented

# 使用示例
augmentor = TextAugmentor()
original_text = "这个产品质量非常好，我很满意"
augmented_texts = augmentor.augment(original_text, n=2)

print(f"原始文本: {original_text}")
print(f"增强后 ({len(augmented_texts)} 条):")
for i, text in enumerate(augmented_texts, 1):
    print(f"  {i}. {text}")

LLM 辅助数据生成

使用大语言模型生成高质量训练数据，特别适合对话、问答类场景：

python

"""
LLM 辅助数据生成 - 用 Claude 生成训练数据
成本：约 ¥0.01/条（Sonnet），质量高于传统增强方法
"""
import anthropic
import json
from typing import List, Dict

client = anthropic.Anthropic()

def generate_training_data(
    domain: str,
    task_type: str,
    num_samples: int = 100,
    examples: List[Dict] = None,
) -> List[Dict]:
    """用 LLM 生成训练数据"""
    
    example_text = ""
    if examples:
        example_text = "\n参考示例：\n" + json.dumps(examples[:3], ensure_ascii=False, indent=2)
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""
请为{domain}领域的{task_type}任务生成 {num_samples} 条训练数据。
{example_text}

要求：
1. 数据多样化，覆盖不同场景
2. 语言自然，符合实际使用习惯
3. 标签准确，无歧义

返回 JSON 数组格式：
[
  {{"input": "输入文本", "output": "期望输出", "label": "分类标签"}},
  ...
]
"""
        }],
    )
    
    return json.loads(response.content[0].text)

# 使用示例：生成客服问答数据
training_data = generate_training_data(
    domain="电商客服",
    task_type="问答",
    num_samples=50,
    examples=[
        {"input": "怎么退货？", "output": "请在订单详情页点击"申请退货"按钮", "label": "退换货"},
        {"input": "物流到哪了？", "output": "请提供订单号，我帮您查询物流状态", "label": "物流查询"},
    ],
)

print(f"生成了 {len(training_data)} 条训练数据")
for item in training_data[:3]:
    print(f"  输入: {item['input'][:30]}...")
    print(f"  输出: {item['output'][:30]}...")
    print(f"  标签: {item['label']}")
    print()

LLM 数据生成 vs 传统增强对比：

方法	成本	质量	多样性	适用场景
同义词替换	低	中	低	文本分类
回译	中	中高	中	通用文本
LLM 生成	中	高	高	对话、问答
人工标注	高	最高	高	所有场景

数据来源：据 Anthropic 2025 研究，LLM 生成的训练数据在对话场景中比传统增强方法准确率高 15-20%。

图像数据增强

python

import torchvision.transforms as transforms
from PIL import Image

# 定义增强管道
transform = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),  # 随机水平翻转
    transforms.RandomRotation(degrees=15),    # 随机旋转
    transforms.RandomResizedCrop(224),        # 随机裁剪
    transforms.ColorJitter(brightness=0.2, contrast=0.2),  # 颜色抖动
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], 
                        std=[0.229, 0.224, 0.225])
])

# 应用增强
image = Image.open("example.jpg")
augmented_image = transform(image)

数据隐私与合规

人机分工：🧑 人定义合规要求 / 🤖 AI 执行脱敏处理

为什么需要关注数据隐私？

法规	适用范围	违规后果	关键要求
GDPR（欧盟）	处理欧盟用户数据	最高年营收 4% 罚款	数据最小化、用户同意、删除权
个人信息保护法（中国）	处理中国用户数据	最高 5000 万元罚款	知情同意、最小必要、本地存储
CCPA（加州）	处理加州用户数据	每次违规 $7,500	用户知情权、删除权

PII 数据脱敏代码

python

"""
PII 数据脱敏工具 - 在数据进入 AI 模型前自动脱敏
据 Microsoft Presidio 2025 文档，PII 脱敏是 AI 应用的必备安全措施
"""
import re
from typing import Dict

class DataAnonymizer:
    """数据脱敏工具"""
    
    # 中国常见 PII 模式
    PII_PATTERNS = {
        "phone": (r"1[3-9]\d{9}", "***手机号***"),
        "id_card": (r"\d{17}[\dXx]", "***身份证***"),
        "email": (r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "***邮箱***"),
        "bank_card": (r"\d{16,19}", "***银行卡***"),
        "name": (r"[一-龥]{2,4}(?=先生|女士|同学|老师)", "***姓名***"),
    }
    
    def anonymize(self, text: str) -> str:
        """脱敏文本中的 PII 信息"""
        result = text
        for pii_type, (pattern, replacement) in self.PII_PATTERNS.items():
            result = re.sub(pattern, replacement, result)
        return result
    
    def detect_pii(self, text: str) -> Dict[str, int]:
        """检测文本中的 PII 数量"""
        findings = {}
        for pii_type, (pattern, _) in self.PII_PATTERNS.items():
            matches = re.findall(pattern, text)
            if matches:
                findings[pii_type] = len(matches)
        return findings

# 使用示例
anonymizer = DataAnonymizer()

text = "请联系张三先生，手机 13812345678，邮箱 zhangsan@example.com"
pii_count = anonymizer.detect_pii(text)
print(f"检测到 PII: {pii_count}")
# 输出: 检测到 PII: {'phone': 1, 'email': 1, 'name': 1}

sanitized = anonymizer.anonymize(text)
print(f"脱敏后: {sanitized}")
# 输出: 脱敏后: 请联系***姓名***先生，手机 ***手机号***，邮箱 ***邮箱***

合规最佳实践：

数据采集前获取用户明确同意
存储时加密，传输时使用 HTTPS
AI 模型训练前自动脱敏 PII
定期审计数据使用情况
建立数据删除机制（用户要求删除时能彻底清除）

OPC 场景下的数据工程 SOP

完整 ETL 管道代码

python

"""
完整 ETL 管道 - 从原始数据到可用数据集
适用于 OPC 场景的小规模数据工程
"""
import pandas as pd
from pathlib import Path
from typing import Dict, List
import json
import hashlib
from datetime import datetime

class ETLPipeline:
    """ETL 管道：Extract → Transform → Load"""
    
    def __init__(self, output_dir: str = "./data/processed"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.log = []
    
    def extract(self, source: str, source_type: str = "csv") -> pd.DataFrame:
        """Extract: 从数据源提取数据"""
        self._log(f"开始提取数据: {source}")
        
        if source_type == "csv":
            df = pd.read_csv(source)
        elif source_type == "json":
            df = pd.read_json(source)
        elif source_type == "excel":
            df = pd.read_excel(source)
        else:
            raise ValueError(f"不支持的数据源类型: {source_type}")
        
        self._log(f"提取完成: {len(df)} 行, {len(df.columns)} 列")
        return df
    
    def transform(self, df: pd.DataFrame, config: Dict) -> pd.DataFrame:
        """Transform: 数据转换和清洗"""
        self._log("开始数据转换")
        
        # 1. 去重
        if config.get("dedup"):
            initial = len(df)
            df = df.drop_duplicates()
            self._log(f"去重: {initial} → {len(df)} 行")
        
        # 2. 缺失值处理
        if config.get("fill_missing"):
            for col, strategy in config["fill_missing"].items():
                if col in df.columns:
                    if strategy == "mean":
                        df[col].fillna(df[col].mean(), inplace=True)
                    elif strategy == "median":
                        df[col].fillna(df[col].median(), inplace=True)
                    elif strategy == "mode":
                        df[col].fillna(df[col].mode()[0], inplace=True)
                    elif strategy == "drop":
                        df.dropna(subset=[col], inplace=True)
                    self._log(f"缺失值处理: {col} ({strategy})")
        
        # 3. 类型转换
        if config.get("type_convert"):
            for col, dtype in config["type_convert"].items():
                if col in df.columns:
                    df[col] = df[col].astype(dtype)
                    self._log(f"类型转换: {col} → {dtype}")
        
        # 4. 过滤
        if config.get("filters"):
            for col, condition in config["filters"].items():
                if col in df.columns:
                    initial = len(df)
                    if condition["op"] == ">":
                        df = df[df[col] > condition["value"]]
                    elif condition["op"] == "<":
                        df = df[df[col] < condition["value"]]
                    elif condition["op"] == "==":
                        df = df[df[col] == condition["value"]]
                    self._log(f"过滤: {col} {condition['op']} {condition['value']}: {initial} → {len(df)}")
        
        # 5. 新增特征
        if config.get("new_features"):
            for new_col, formula in config["new_features"].items():
                df[new_col] = eval(formula, {"df": df, "pd": pd})
                self._log(f"新增特征: {new_col}")
        
        self._log(f"转换完成: {len(df)} 行, {len(df.columns)} 列")
        return df
    
    def load(self, df: pd.DataFrame, filename: str, format: str = "csv"):
        """Load: 保存处理后的数据"""
        output_path = self.output_dir / filename
        
        if format == "csv":
            df.to_csv(output_path, index=False)
        elif format == "json":
            df.to_json(output_path, orient="records", force_ascii=False, indent=2)
        elif format == "parquet":
            df.to_parquet(output_path, index=False)
        
        # 计算数据指纹
        fingerprint = hashlib.md5(pd.util.hash_pandas_object(df).values.tobytes()).hexdigest()
        
        self._log(f"数据已保存: {output_path}")
        self._log(f"数据指纹: {fingerprint}")
        
        # 保存处理日志
        log_path = self.output_dir / f"{filename}.log.json"
        with open(log_path, "w", encoding="utf-8") as f:
            json.dump({
                "timestamp": datetime.now().isoformat(),
                "filename": filename,
                "rows": len(df),
                "columns": len(df.columns),
                "fingerprint": fingerprint,
                "steps": self.log,
            }, f, ensure_ascii=False, indent=2)
        
        return output_path
    
    def _log(self, message: str):
        """记录日志"""
        entry = f"[{datetime.now().strftime('%H:%M:%S')}] {message}"
        self.log.append(entry)
        print(entry)

# 使用示例
pipeline = ETLPipeline(output_dir="./data/processed")

# Extract
df = pipeline.extract("./data/raw/customers.csv", source_type="csv")

# Transform
config = {
    "dedup": True,
    "fill_missing": {
        "age": "median",
        "city": "mode",
    },
    "type_convert": {
        "age": "int",
        "signup_date": "datetime64[ns]",
    },
    "filters": {
        "age": {"op": ">", "value": 0},
    },
}

df_clean = pipeline.transform(df, config)

# Load
pipeline.load(df_clean, "customers_clean.csv", format="csv")

小团队数据工程流程

每日 SOP

时间	任务	工具	产出
09:00	检查数据质量报告	自动脚本	质量报告
09:30	处理异常数据	pandas	清洗日志
10:00	标注新样本	Label Studio	标注数据
14:00	审核标注质量	人工抽检	审核报告
16:00	更新数据版本	DVC	版本标签

每周 SOP

任务	频率	说明
数据源更新	每周一	检查数据源是否有新数据
质量趋势分析	每周五	分析本周数据质量变化
标注规范更新	按需	根据问题更新标注指南
数据集版本发布	每周五	发布本周数据集版本

成本控制建议

策略	效果	实施方法
优先使用公开数据集	降低采集成本	先用公开数据验证，再决定是否自采
AI 辅助标注	降低标注成本 60-70%	先用模型预标注，人工审核修正
主动学习	降低标注量 50%	只标注最有价值的样本
增量更新	降低存储成本	只存储变更部分

常见问题

问题	原因	解决方案
模型效果差	数据质量低	先清洗数据，再优化模型
标注不一致	标注规范不清晰	编写详细标注文档，定期校准
数据泄露	训练集包含测试数据	严格划分数据集，检查数据来源
过拟合	数据量不足	数据增强或收集更多数据
标注成本高	样本选择不当	使用主动学习，优先标注高价值样本
数据版本混乱	缺乏版本管理	使用 DVC 或类似工具
增强数据质量差	增强方法不当	人工审核增强结果，调整策略
向量检索不准	Embedding 模型不适合	测试不同 Embedding 模型，选择领域最优
数据采集被封	请求频率过高	添加延时、使用代理池、遵守 robots.txt

下一步

完成数据准备后，进入阶段 3：模型选择与训练

参考与延伸

[1] HuggingFace. "Datasets Documentation"（2025）— 数据集工具文档

[2] DVC. "Data Version Control"（2025）— 数据版本管理

[3] Google. "Data Validation Guide"（2025）— TensorFlow 数据验证工具

[4] Label Studio. "Documentation"（2025）— 开源数据标注平台

[5] Scikit-learn. "Model Selection"（2025）— 数据集划分与交叉验证

[6] NLPAug. "Text Augmentation"（2025）— 文本数据增强库

[7] AWS. "Data Engineering Best Practices"（2025）— 数据工程最佳实践

[8] Anthropic. "Data Generation for Fine-tuning"（2025）— LLM 辅助数据生成最佳实践

[9] Argilla. "Documentation"（2025）— LLM 数据标注平台

[10] Microsoft. "Presidio: Data Protection SDK"（2025）— PII 检测和脱敏工具，支持 30+ 种 PII 类型

[11] NIST. "AI Risk Management Framework"（2025）— AI 风险管理框架，包含数据治理指南

2.14 数据工程 ​

什么是数据工程？ ​

核心定义 ​

为什么占 60-80% 工作量？ ​

人机分工 ​

数据工程四步流程 ​

Step 1：数据采集 ​

数据来源 ​

API 抓取示例 ​

网页爬取示例 ​

推荐数据集平台 ​

Step 2：数据清洗 ​

核心任务 ​

完整清洗流程 ​

数据清洗代码示例 ​

Step 3：数据标注 ​

标注流程 ​

标注策略 ​

标注工具使用示例 ​

Label Studio 安装与配置 ​

Label Studio Python API ​

主动学习示例 ​

推荐标注工具 ​

Step 4：数据版本管理 ​

为什么需要数据版本管理？ ​

DVC 使用示例 ​

安装与初始化 ​

数据版本控制 ​

DVC Pipeline ​

推荐工具 ​

数据质量评估 ​

评估指标 ​

数据质量检查流程 ​

质量评估代码 ​

数据集划分策略 ​

划分比例 ​

划分代码示例 ​

注意事项 ​

数据增强 ​

为什么需要数据增强？ ​

数据增强策略选择流程 ​

文本数据增强 ​

LLM 辅助数据生成 ​

图像数据增强 ​

数据隐私与合规 ​

为什么需要关注数据隐私？ ​

PII 数据脱敏代码 ​

OPC 场景下的数据工程 SOP ​

完整 ETL 管道代码 ​

小团队数据工程流程 ​

每日 SOP ​

每周 SOP ​

成本控制建议 ​

常见问题 ​

下一步 ​

参考与延伸 ​

2.14 数据工程

什么是数据工程？

核心定义

为什么占 60-80% 工作量？

人机分工

数据工程四步流程

Step 1：数据采集

数据来源

API 抓取示例

网页爬取示例

推荐数据集平台

Step 2：数据清洗

核心任务

完整清洗流程

数据清洗代码示例

Step 3：数据标注

标注流程

标注策略

标注工具使用示例

Label Studio 安装与配置

Label Studio Python API

主动学习示例

推荐标注工具

Step 4：数据版本管理

为什么需要数据版本管理？

DVC 使用示例

安装与初始化

数据版本控制

DVC Pipeline

推荐工具

数据质量评估

评估指标

数据质量检查流程

质量评估代码

数据集划分策略

划分比例

划分代码示例

注意事项

数据增强

为什么需要数据增强？

数据增强策略选择流程

文本数据增强

LLM 辅助数据生成

图像数据增强

数据隐私与合规

为什么需要关注数据隐私？

PII 数据脱敏代码

OPC 场景下的数据工程 SOP

完整 ETL 管道代码

小团队数据工程流程

每日 SOP

每周 SOP

成本控制建议

常见问题

下一步

参考与延伸