clipper/pyspark-container

Clipper PySpark容器

镜像概述

Clipper是一款介于用户应用与各类机器学习模型/框架之间的预测服务系统，本容器专注于将PySpark模型部署到Clipper，帮助用户快速实现模型的生产级服务化，降低应用与模型的集成门槛。

核心功能

1. 简化集成：提供统一REST接口，支持多框架模型的预测与反馈，让产品团队轻松接入机器学习能力。
2. 简化部署：复用模型开发时的工具与库进行实时预测，减少部署过程中的常见bug，提升数据科学家的工作效率。
3. 性能优化：通过自适应批处理、缓存和拖慢请求缓解技术，提升吞吐量并保证毫秒级延迟，降低运维团队的负担。
4. 精度提升：采用先进的bandit算法和集成方法，智能选择组合预测结果，实现实时个性化，提升用户体验。

使用场景

• 需要将PySpark模型快速上线的生产环境场景。
• 对预测延迟和吞吐量有严格要求的用户应用。
• 需整合多种机器学习框架模型的系统。

配置说明与部署示例

前提条件

• 已安装Docker。
• 具备Python 2运行环境。

安装Clipper

从GitHub直接安装：

sh
pip install git+https://github.com/ucbrise/clipper.git@develop#subdirectory=clipper_admin

或克隆仓库后本地安装：

sh
pip install -e </path/to/clipper_repo>/clipper_admin

启动本地Clipper集群

打开Python交互环境：

sh
python
# 或使用iPython
conda install ipython
ipython

执行以下代码：

py
from clipper_admin import ClipperConnection, DockerContainerManager
clipper_conn = ClipperConnection(DockerContainerManager())

# 启动Clipper（首次运行会下载依赖容器，耗时可能较长）
clipper_conn.start_clipper()

# 注册应用"hello-world"，创建预测端点http://localhost:1337/hello-world/predict
clipper_conn.register_application(name="hello-world", input_type="doubles", default_output="-1.0", slo_micros=100000)

# 定义特征求和模型
def feature_sum(xs):
    return [str(sum(x)) for x in xs]

# 部署模型到Clipper
from clipper_admin.deployers import python as python_deployer
python_deployer.deploy_python_closure(clipper_conn, name="sum-model", version=1, input_type="doubles", func=feature_sum)

# 链接应用与模型
clipper_conn.link_model_to_app(app_name="hello-world", model_name="sum-model")

查询预测结果

使用cURL：

sh
curl -X POST --header "Content-Type:application/json" -d '{"input": [1.1, 2.2, 3.3]}' 127.0.0.1:1337/hello-world/predict

或Python：

py
import requests, json, numpy as np
headers = {"Content-type": "application/json"}
response = requests.post("http://localhost:1337/hello-world/predict", headers=headers, data=json.dumps({"input": list(np.random.random(10))}))
print(response.json())

清理资源

py
# 若已关闭之前的会话，重新连接集群
from clipper_admin import ClipperConnection, DockerContainerManager
clipper_conn = ClipperConnection(DockerContainerManager())
clipper_conn.connect()

# 停止所有Clipper容器
clipper_conn.stop_all()