1. 用纯Python构建生产级机器学习系统的核心组件

作为一名在Etsy机器学习平台团队工作多年的工程师，我深知学习生产级机器学习系统的困难。大多数教程要么停留在理论层面，要么要求你掌握一整套复杂的工具链（Kubernetes、Docker、Airflow等）。今天，我将用纯Python代码演示生产系统中常见的6个核心组件，让你在本地就能理解它们的运作原理。

重要提示：本文所有代码都是教学演示性质，真实生产系统需要考虑分布式计算、容错机制等更多因素。但这些代码展示了每个组件的核心思想，你可以基于此逐步扩展。

2. 基础环境准备

2.1 最小化依赖配置

我们只需要3个基础库：

pip install pandas scikit-learn pytest

2.2 示例数据集说明

使用合成的用户行为数据集演示，包含两个特征和一个二分类标签：

import pandas as pd
url = "https://raw.githubusercontent.com/kylegallatin/components-of-an-ml-system/main/data/user_data.csv"
user_data = pd.read_csv(url, index_col=0)

print(user_data.head())
"""
                                feature_1  feature_2  target
user_id                                                     
a27b09ee-0cf8-11ed-899e-b29c4abd48f4        0.548814  0.592040       1
a27b09f2-0cf8-11ed-899e-b29c4abd48f4        0.715189  0.844265       1
a27b09f4-0cf8-11ed-899e-b29c4abd48f4        0.602763  0.857945       0
"""

3. 特征存储(Feature Store)实现

3.1 为什么需要特征存储？

生产环境中，特征可能被多个模型共享，且需要保证训练和服务时特征计算的一致性。以下是简化实现：

from typing import Dict, Callable

class SuperSimpleFeatureStore:
    def __init__(self, data: pd.DataFrame):
        self.data = data.to_dict('index')
        self.feature_funcs = {}
    
    def register_feature(self, name: str, func: Callable):
        """动态注册新特征"""
        self.feature_funcs[name] = func
        # 立即应用新特征到现有数据
        for user_id, features in self.data.items():
            features[name] = func(features)
    
    def get_user_feature(self, user_id: str) -> Dict:
        """实时特征获取"""
        features = self.data[user_id].copy()
        for name, func in self.feature_funcs.items():
            features[name] = func(features)
        return features

# 使用示例
feature_store = SuperSimpleFeatureStore(user_data)

# 定义特征工程函数
def feature_1_squared(feature_dict: Dict) -> float:
    return feature_dict["feature_1"] ** 2

feature_store.register_feature("feature_1_squared", feature_1_squared)
print(feature_store.get_user_feature("a27b09ee-0cf8-11ed-899e-b29c4abd48f4"))

3.2 生产级考量

实际系统中需要：

• 特征版本控制
• 离线/在线特征计算一致性
• 分布式存储支持
• 低延迟查询优化

4. 实验跟踪与模型注册

4.1 元数据记录实现

import csv
from datetime import datetime
from sklearn.metrics import accuracy_score

class ExperimentTracker:
    def __init__(self, log_file: str = "experiments.csv"):
        self.log_file = log_file
        self._init_log_file()
    
    def _init_log_file(self):
        with open(self.log_file, 'w') as f:
            writer = csv.writer(f)
            writer.writerow([
                'timestamp', 'model_type', 'params', 
                'train_acc', 'test_acc', 'model_path'
            ])
    
    def log_experiment(self, model, X_train, X_test, y_train, y_test):
        timestamp = datetime.now().isoformat()
        model_type = type(model).__name__
        params = str(model.get_params())
        
        train_pred = model.predict(X_train)
        test_pred = model.predict(X_test)
        
        train_acc = accuracy_score(y_train, train_pred)
        test_acc = accuracy_score(y_test, test_pred)
        
        model_path = f"models/{timestamp}.pkl"
        joblib.dump(model, model_path)
        
        with open(self.log_file, 'a') as f:
            writer = csv.writer(f)
            writer.writerow([
                timestamp, model_type, params,
                train_acc, test_acc, model_path
            ])

# 使用示例
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import joblib

X = user_data[['feature_1', 'feature_2']]
y = user_data['target']
X_train, X_test, y_train, y_test = train_test_split(X, y)

tracker = ExperimentTracker()
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)
tracker.log_experiment(model, X_train, X_test, y_train, y_test)

4.2 模型注册表示例

class ModelRegistry:
    def __init__(self):
        self.models = {}
        self.current_model = None
    
    def register_model(self, name: str, path: str):
        self.models[name] = {
            'path': path,
            'timestamp': datetime.now().isoformat()
        }
        self.current_model = name
    
    def load_model(self, name: str):
        return joblib.load(self.models[name]['path'])

registry = ModelRegistry()
registry.register_model("rf_v1", "models/2023-07-01T12:00:00.pkl")

5. 自动化训练流水线

5.1 简易训练调度器

import time
from schedule import Scheduler

class TrainingPipeline:
    def __init__(self, data: pd.DataFrame, interval: int = 60):
        self.data = data
        self.scheduler = Scheduler()
        self.interval = interval
    
    def train_job(self):
        print(f"{datetime.now()} - Starting training job...")
        X, y = self.data[['feature_1', 'feature_2']], self.data['target']
        model = RandomForestClassifier()
        model.fit(X, y)
        
        model_path = f"models/{datetime.now().isoformat()}.pkl"
        joblib.dump(model, model_path)
        print(f"Model saved to {model_path}")
    
    def run_continuously(self):
        self.scheduler.every(self.interval).seconds.do(self.train_job)
        while True:
            self.scheduler.run_pending()
            time.sleep(1)

# 启动后台训练
pipeline = TrainingPipeline(user_data)
pipeline.run_continuously()  # 实际运行需在后台进程

5.2 生产级改进方向

• 使用Celery或Airflow实现分布式任务调度
• 添加数据新鲜度检查
• 实现模型性能自动评估
• 失败重试机制

6. 持续集成与测试

6.1 机器学习测试样例

创建 test_model.py ：

import pytest
from sklearn.pipeline import Pipeline

def test_model_loading():
    model = joblib.load("models/latest.pkl")
    assert isinstance(model, (RandomForestClassifier, Pipeline))

def test_feature_store():
    fs = SuperSimpleFeatureStore(user_data)
    assert len(fs.get_user_feature(next(iter(user_data.index)))) >= 2

def test_data_quality():
    assert not user_data.isnull().any().any()
    assert (user_data['target'].isin([0, 1])).all()

6.2 集成到Git Hook

创建 .git/hooks/pre-commit ：

#!/bin/sh
pytest test_model.py || exit 1

7. 模型服务与监控

7.1 简易预测服务

from fastapi import FastAPI
import uvicorn

app = FastAPI()
registry = ModelRegistry()

@app.get("/predict/{user_id}")
def predict(user_id: str):
    features = feature_store.get_user_feature(user_id)
    model = registry.load_model(registry.current_model)
    
    start_time = time.time()
    pred = model.predict([[features['feature_1'], features['feature_2']]])
    latency = time.time() - start_time
    
    return {
        "prediction": int(pred[0]),
        "model_version": registry.current_model,
        "latency_ms": round(latency * 1000, 2)
    }

# 启动服务：uvicorn main:app --reload

7.2 性能监控实现

class PredictionMonitor:
    def __init__(self, window_size: int = 100):
        self.latencies = []
        self.window_size = window_size
    
    def record(self, latency: float):
        self.latencies.append(latency)
        if len(self.latencies) > self.window_size:
            self.latencies.pop(0)
    
    def mean_latency(self) -> float:
        return sum(self.latencies) / len(self.latencies) if self.latencies else 0
    
    def percentile(self, p: int) -> float:
        return np.percentile(self.latencies, p) if self.latencies else 0

monitor = PredictionMonitor()

8. 系统扩展建议

1. 数据管道 ：添加Apache Beam或Spark实现批流一体处理
2. 特征回填 ：实现时间旅行(Time Travel)功能保证历史特征一致性
3. 模型监控 ：添加数据漂移和模型性能衰减检测
4. 服务高可用 ：使用Redis缓存模型和特征数据

这些组件构成了机器学习系统的基础骨架。在我的实际工作中，每个组件都需要考虑：

• 水平扩展能力
• 故障恢复机制
• 安全审计日志
• 资源使用监控

建议从这个小系统开始，逐步添加复杂度。比如先用Redis替换内存存储，再用Kafka实现事件驱动架构，最后引入Kubernetes管理容器化服务。记住，生产系统的核心不是工具的复杂度，而是可靠性和可维护性。