ML 元数据管理

概述

使用 MLMD 跟踪 ML 工件、血缘和来源。本指南涵盖生产级 ML 系统的实际实现。

为什么这在 MLOps 中很重要

现代 ML 系统需要严格的操作实践：

设置

bash
安装所需工具
pip install mlmd mlflow pandas numpy scikit-learn
或使用 Docker
docker pull python:3.11-slim

核心实现

python
import os
import json
import logging
from datetime import datetime
from pathlib import Path
logger = logging.getLogger(__name__)
class MLMetadataManagement:
    """
    ML 元数据管理实现。
    
    处理：血缘
    工具：mlmd
    """
    
    def __init__(self, config: dict = None):
        self.config = config or self._default_config()
        self._setup()
    
    def _default_config(self) -> dict:
        return {
            "tool": "mlmd",
            "environment": os.getenv("ENVIRONMENT", "development"),
            "log_level": "INFO",
        }
    
    def _setup(self):
        """初始化 mlmd 连接和资源。"""
        logging.basicConfig(level=self.config.get("log_level", "INFO"))
        logger.info(f"使用配置初始化 ML 元数据管理：{self.config}")
    
    def run(self, **kwargs) -> dict:
        """执行血缘跟踪。"""
        start = datetime.utcnow()
        
        try:
            result = self._execute(**kwargs)
            
            elapsed = (datetime.utcnow() - start).total_seconds()
            logger.info(f"ML 元数据管理完成，耗时 {elapsed:.2f} 秒")
            
            return {
                "status": "success",
                "result": result,
                "elapsed_seconds": elapsed
            }
        
        except Exception as e:
            logger.error(f"ML 元数据管理失败：{e}")
            return {
                "status": "failed", 
                "error": str(e)
            }
    
    def _execute(self, **kwargs) -> dict:
        """核心血缘逻辑。可重写以自定义。"""
        return {"completed": True, "tool": "mlmd"}
配置
config = {
    "tool": "mlmd",
    "tracking_uri": os.getenv("MLFLOW_TRACKING_URI", "http://localhost:5000"),
    "artifact_root": "./artifacts",
}
初始化
processor = MLMetadataManagement(config)
result = processor.run()
print(json.dumps(result, indent=2))

MLMD 集成

python
用于血缘跟踪的特定 mlmd 集成
import subprocess
def setup_mlmd():
    """配置 mlmd 进行血缘跟踪。"""
    
    # 初始化项目
    print(f"正在设置 mlmd 进行血缘跟踪...")
    
    # 示例配置
    config = {
        "project": "my-ml-project",
        "tool": "mlmd",
        "specialty": "lineage",
        "version": "1.0.0"
    }
    
    # 保存配置
    Path(".mlmd").mkdir(exist_ok=True)
    with open(f".mlmd/config.json", "w") as f:
        json.dump(config, f, indent=2)
    
    print(f"mlmd 已配置用于血缘跟踪")
    return config
config = setup_mlmd()

监控与告警

python
from dataclasses import dataclass
import time
class MLOpsMonitor:
    """监控血缘指标。"""
    
    def __init__(self):
        self.metrics: list[MetricSnapshot] = []
        self.thresholds = {
            "error_rate": 0.05,
            "latency_p99_ms": 1000,
            "data_drift_score": 0.3
        }
    
    def record(self, metric: str, value: float, labels: dict = None):
        snapshot = MetricSnapshot(
            timestamp=time.time(),
            metric_name=metric,
            value=value,
            labels=labels or {}
        )
        self.metrics.append(snapshot)
        self._check_threshold(metric, value)
    
    def _check_threshold(self, metric: str, value: float):
        threshold = self.thresholds.get(metric)
        if threshold and value > threshold:
            logger.warning(f"告警：{metric}={value:.3f} 超过阈值 {threshold}")
monitor = MLOpsMonitor()

CI/CD 集成

yaml
.github/workflows/ml-pipeline.yml
name: ML Pipeline
on:
  push:
    paths: ['src/', 'data/']
jobs:
  train-and-evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run lineage
        run: python -m src.ml_metadata_management
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}
      
      - name: Check model quality
        run: python -m src.validate_model

最佳实践

资源