一、项目概述
1.1 项目目标
开发一套支持水平拓展的传统机器学习服务，部署于K8s环境，实现以下核心目标：
支持多类型传统机器学习模型（有监督：随机森林、逻辑回归；无监督：DBSCAN、K-Means；预处理：PCA等）的统一调用。
实现任务分布式调度，将训练/预测任务下发至多个Worker节点，避免单一节点负载过高。
基于MinIO实现分布式数据/模型存储，基于Redis实现任务队列、模型缓存与状态管理。
设计统一RESTful API，以
/<modelType>/<method>格式适配任意模型的各类操作（fit、predict、fit_predict等），提升接口可扩展性与可维护性。支持K8s环境下的弹性扩缩容，适配任务量动态变化，保障服务稳定性与资源利用率。
1.2 核心技术栈
| 技术类别 | 核心技术 | 作用说明 |
|---|---|---|
| 容器编排 | Kubernetes（K8s） | 服务部署、容器调度、弹性扩缩容、负载均衡 |
| 机器学习框架 | Scikit-learn | 传统机器学习模型（分类、回归、聚类、预处理）实现 |
| 任务调度 | Celery + Redis | 异步任务分发、任务队列管理、任务状态缓存 |
| 分布式存储 | MinIO | 训练数据、模型文件、任务结果的分布式存储与管理 |
| API服务 | FastAPI + Uvicorn | 提供统一RESTful API，支持参数校验、异步响应 |
| 进程管理 | Supervisord | 在Worker容器内管理Celery Worker与FastAPI服务进程 |
二、系统架构设计
2.1 整体架构分层
系统采用分层架构设计，各层独立解耦，支持水平拓展，整体分为5层：
接入层：基于K8s Service + Ingress（或Traefik网关）实现请求路由与负载均衡，将外部API请求随机分发至任意Worker节点，避免单点压力。
API服务层：部署于Worker节点，基于FastAPI实现统一接口定义、参数校验、请求分发，对接核心业务逻辑。
任务调度层：Celery作为任务调度器，Redis作为消息代理（任务队列）与结果后端，接收API层传递的任务，根据节点负载下发至空闲Worker。
计算层：无状态Worker节点（K8s Deployment部署），执行模型训练、预测、聚类等核心任务，支持基于HPA实现弹性扩缩容。
存储层：MinIO分布式对象存储集群存储数据与模型，Redis集群存储任务状态、模型缓存、热点结果，保障数据可靠性与访问性能。
2.2 核心组件交互流程
2.2.1 有监督模型训练流程（如随机森林）
客户端通过统一API（
/rf/fit）提交训练请求，携带数据路径、特征列名、模型参数等信息。网关将请求路由至任意Worker节点的FastAPI服务，API层校验参数后提交任务至Redis队列。
Celery Worker从Redis队列拉取任务，从MinIO下载训练数据并加载。
执行模型训练（fit），训练完成后将模型上传至MinIO，同时更新Redis中任务状态为“成功”。
模型缓存至Worker节点内存（按配置设置过期时间），API层返回任务ID、模型存储路径等结果。
2.2.2 无监督模型聚类流程（如DBSCAN）
客户端通过统一API（
/dbscan/fit_predict）提交聚类请求，携带数据路径、特征列名、DBSCAN参数。FastAPI校验参数后提交异步任务至Redis队列，Celery Worker拉取任务并从MinIO加载数据。
执行fit_predict（训练+聚类一体），生成聚类结果与统计信息。
聚类结果保存至MinIO，API层返回任务ID、结果路径、聚类统计（类别数量、噪声点数量）。
2.2.3 模型预测流程（复用缓存）
客户端通过统一API（
/rf/predict）提交预测请求，携带模型名称、数据路径、特征列名。Worker节点优先从内存缓存加载模型（缓存未命中/过期则从MinIO下载并更新缓存）。
加载MinIO中的预测数据，执行预测并将结果缓存至Redis（按需）。
API层返回预测结果、结果存储路径。
三、核心组件实现
3.1 依赖配置（requirements.txt）
celery==5.3.4 redis==5.0.1 minio==7.2.0 scikit-learn==1.3.0 xgboost==2.0.0 pandas==2.1.0 numpy==1.25.2 joblib==1.3.2 fastapi==0.104.1 uvicorn==0.24.0 pydantic==2.4.2 supervisor==4.2.53.2 配置文件（config.py）
统一管理Redis、MinIO、Celery连接配置，支持通过环境变量适配K8s部署：
importos# Redis 配置（对应K8s中Redis Service地址）REDIS_HOST=os.getenv("REDIS_HOST","redis-service.ml-service.svc.cluster.local")REDIS_PORT=int(os.getenv("REDIS_PORT",6379))REDIS_DB=0# MinIO 配置MINIO_HOST=os.getenv("MINIO_HOST","minio-service.ml-service.svc.cluster.local")MINIO_PORT=int(os.getenv("MINIO_PORT",9000))MINIO_ACCESS_KEY=os.getenv("MINIO_ACCESS_KEY","mlminioadmin")MINIO_SECRET_KEY=os.getenv("MINIO_SECRET_KEY","mlminiopassword")MINIO_BUCKET="ml-models"# 存储训练数据、模型、结果的桶# Celery 配置CELERY_BROKER_URL=f"redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}"CELERY_RESULT_BACKEND=f"redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}"3.3 统一请求/返回模型（schemas.py）
基于Pydantic定义标准化参数结构，实现请求校验与格式统一：
frompydanticimportBaseModel,FieldfromtypingimportList,Optional,Dict,Any# 数据信息子模型classDataInfo(BaseModel):data_path:str=Field(...,description="MinIO中的数据文件路径（如 data/train.csv）")feature_cols:List[str]=Field(...,description="特征列名列表")label_col:Optional[str]=Field(None,description="标签列名（仅有监督模型fit时需要）")result_name:Optional[str]=Field(None,description="模型/结果保存名称（默认自动生成）")# 执行配置子模型classExecConfig(BaseModel):cache_model:Optional[bool]=Field(True,description="是否缓存模型")expire_seconds:Optional[int]=Field(3600,description="模型缓存过期时间（秒）")# 统一请求模型classMLRequest(BaseModel):data_info:DataInfo model_params:Optional[Dict[str,Any]]=Field({},description="模型超参数（无则使用默认值）")exec_config:Optional[ExecConfig]=Field(ExecConfig(),description="执行配置")# 统一返回模型classMLResponse(BaseModel):code:int=Field(200,description="状态码（200成功，500失败）")msg:str=Field("操作成功",description="提示信息")data:Optional[Dict[str,Any]]=Field({},description="返回数据（任务ID、结果路径等）")3.4 模型注册表与分发器（model_registry.py）
核心组件，维护模型与方法的映射关系，实现动态分发，支持快速扩展模型：
fromtypingimportDict,Callable,Listfromsklearn.ensembleimportRandomForestClassifierfromsklearn.clusterimportDBSCAN,KMeansfromsklearn.decompositionimportPCAfromsklearn.linear_modelimportLogisticRegression# 模型注册表：{modelType: (模型类, 支持的方法列表, 模型类型说明)}MODEL_REGISTRY:Dict[str,tuple]={"rf":(RandomForestClassifier,["fit","predict","predict_proba"],"有监督分类（随机森林）"),"dbscan":(DBSCAN,["fit_predict"],"无监督聚类（DBSCAN）"),"pca":(PCA,["fit","transform","fit_transform"],"特征降维（PCA）"),"lr":(LogisticRegression,["fit","predict","predict_proba"],"有监督分类（逻辑回归）"),"kmeans":(KMeans,["fit","predict","fit_predict"],"无监督聚类（K-Means）")}# 方法映射表：接口method映射到sklearn模型方法METHOD_MAP:Dict[str,str]={"fit":"fit","predict"
