从Model-Centric到Data-Centric：MLOps实践中的范式转变与落地指南

1次阅读

共计 3559 个字符，预计需要花费 9 分钟才能阅读完成。

在机器学习项目实践中，我们常常看到这样的现象：相同的模型架构，在不同质量的数据集上表现差异巨大。让我们来看两个真实案例：

某电商推荐系统项目中，使用相同的 BERT 模型架构，在清洗后的高质量用户行为数据上 AUC 达到 0.92，而在原始数据上仅为 0.78。差异主要来自数据中的噪声和缺失值。
一个医疗影像分类任务中，ResNet50 模型在专业标注员标注的数据集上准确率为 93%，而在众包标注数据上只有 85%。标注质量直接影响模型性能上限。

这些案例清晰地表明，仅优化模型架构 (model-centric) 已经无法满足生产需求，我们必须转向关注数据质量 (data-centric) 的 AI 开发范式。

建立量化评估是 Data-Centric AI 的第一步。以下是核心指标和实现代码：

from typing import Dict, Any
import pandas as pd
import numpy as np

def calculate_data_quality(df: pd.DataFrame) -> Dict[str, Any]:
    """
    计算数据集质量指标

    Args:
        df: 待评估的 DataFrame

    Returns:
        包含各项指标的字典
    """metrics = {'missing_rate': df.isnull().mean().mean(),'duplicate_rate': df.duplicated().mean(),'outlier_rate': _calculate_outlier_rate(df),'class_imbalance': _calculate_class_imbalance(df)
    }
    return metrics

def _calculate_outlier_rate(df: pd.DataFrame, threshold: float = 3.0) -> float:
    """计算数值特征的异常值比例"""
    numeric_cols = df.select_dtypes(include=np.number).columns
    z_scores = (df[numeric_cols] - df[numeric_cols].mean()) / df[numeric_cols].std()
    return (np.abs(z_scores) > threshold).mean().mean()

关键指标说明：

缺失率(missing_rate)：数据集中缺失值的比例
重复率(duplicate_rate)：重复样本的比例
异常值率(outlier_rate)：超出 3σ 范围的数值比例
类别不平衡度(class_imbalance)：分类任务中少数类的占比

基于 Airflow 的自动化清洗流水线示例：

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator


def clean_data(raw_data_path: str, clean_data_path: str):
    """执行数据清洗的核心逻辑"""
    df = pd.read_parquet(raw_data_path)

    # 处理缺失值
    df = df.fillna(method='ffill').fillna(method='bfill')

    # 去除重复样本
    df = df.drop_duplicates()

    # 保存清洗结果
    df.to_parquet(clean_data_path)


default_args = {
    'owner': 'ml_team',
    'start_date': datetime(2023, 1, 1)
}

with DAG(
    'data_cleaning_pipeline',
    default_args=default_args,
    schedule_interval='@daily'
) as dag:

    cleaning_task = PythonOperator(
        task_id='clean_raw_data',
        python_callable=clean_data,
        op_kwargs={
            'raw_data_path': '/data/raw/daily_data.parquet',
            'clean_data_path': '/data/clean/cleaned_data.parquet'
        }
    )

流水线设计要点：

模块化设计：每个清洗步骤独立可配置
数据溯源：保留原始数据和清洗步骤的元数据
监控告警：对清洗后的数据质量进行自动校验

主动学习 (Active Learning) 能显著提升标注效率。核心算法公式：

$$
\text{Uncertainty}(x) = 1 – \max_y P(y|x;\theta)
$$

其中 $P(y|x;\theta)$ 是模型对样本 $x$ 的预测概率分布。我们优先标注模型最不确定的样本。

Python 实现示例：

from sklearn.ensemble import RandomForestClassifier
from typing import List, Tuple


def active_learning_sampling(
    model: RandomForestClassifier, 
    unlabeled_data: pd.DataFrame,
    batch_size: int
) -> Tuple[pd.DataFrame, List[int]]:
    """
    主动学习样本选择

    Args:
        model: 当前训练好的模型
        unlabeled_data: 未标注数据
        batch_size: 需要选择的样本数

    Returns:
        (selected_samples, uncertain_indices)
    """
    probs = model.predict_proba(unlabeled_data)
    uncertainties = 1 - np.max(probs, axis=1)
    selected_indices = np.argsort(uncertainties)[-batch_size:]
    return unlabeled_data.iloc[selected_indices], selected_indices

推荐使用 DVC 管理数据版本：

# 初始化 DVC
$ dvc init

# 添加数据集
$ dvc add data/raw/dataset.csv

# 提交变更
$ git add .
$ git commit -m "Track dataset v1.0"

使用群体稳定性指数 (PSI) 检测数据分布变化：

$$
\text{PSI} = \sum (\text{实际比例} – \text{预期比例}) \times \ln\left(\frac{\text{实际比例}}{\text{预期比例}}\right)
$$

Python 实现：

def calculate_psi(expected: np.array, actual: np.array, bins: int = 10) -> float:
    """计算两个分布的 PSI 值"""
    breakpoints = np.linspace(0, 1, bins)
    expected_percents = np.histogram(expected, breakpoints)[0] / len(expected)
    actual_percents = np.histogram(actual, breakpoints)[0] / len(actual)
    return np.sum((actual_percents - expected_percents) * 
                 np.log(actual_percents / expected_percents))

PSI 解读：