MLlib 机器学习基础

概述

MLlib 是 Spark 的机器学习库，提供 ML Pipeline 统一特征工程与模型训练，覆盖分类、回归、聚类、协同过滤等主流算法。本文讲清 Pipeline 机制、四大类算法选型与用法，以及模型保存/加载。

---

一、MLlib 定位

特性	说明
分布式	基于 DataFrame 分布式训练
统一 API	Pipeline 串起全流程
内置算法	分类/回归/聚类/推荐
集成	与 Spark SQL 无缝衔接

对比	适用
MLlib	中大规模、标准算法
专用库（XGBoost）	更强单模型
深度学习	换 TensorFlow/PyTorch

---

二、ML Pipeline

2.1 核心概念

组件	作用
Transformer	转换器（特征变换，有 fit 后 transform）
Estimator	估计器（算法模型，fit 训练出模型）
Pipeline	阶段链（Transformer 和 Estimator 串联）

Pipeline = [Tokenizer, HashingTF, LogisticRegression]
  fit(trainData) → PipelineModel
  transform(testData) → 预测结果

2.2 代码示例

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}

val tokenizer = new Tokenizer().setInputCol("text").setOutputCol("words")
val hashingTF = new HashingTF()
  .setInputCol("words").setOutputCol("features").setNumFeatures(1000)
val lr = new LogisticRegression().setMaxIter(10)

val pipeline = new Pipeline().setStages(Array(tokenizer, hashingTF, lr))
val model = pipeline.fit(training)
model.transform(test).select("prediction").show()

---

三、特征工程

3.1 常用 Transformer

组件	作用
Tokenizer	分词
StringIndexer	类别转索引
OneHotEncoder	独热编码
VectorAssembler	多列合并特征向量
StandardScaler	标准化
MinMaxScaler	归一化
PCA	降维

3.2 特征流程

原始列 → StringIndexer → OneHotEncoder ┐
原始列 → VectorAssembler ──────────────┴→ features → 模型

val assembler = new VectorAssembler()
  .setInputCols(Array("age", "income", "city_index"))
  .setOutputCol("features")

---

四、分类算法

4.1 算法对比

算法	特点	适用
逻辑回归	线性、可解释、快	二分类基线
决策树	非线性、可解释	中小特征
随机森林	集成、稳、抗过拟合	分类默认选择
GBT	梯度提升、精度高	追求精度
SVM	线性/核	高维小样本

4.2 使用

val rf = new RandomForestClassifier()
  .setNumTrees(100).setMaxDepth(10)
  .setLabelCol("label").setFeaturesCol("features")

4.3 评估

指标	说明
accuracy	准确率
precision/recall	精确率/召回率
f1	综合指标
AUC	排序能力

val evaluator = new BinaryClassificationEvaluator()
  .setMetricName("areaUnderROC")

---

五、回归算法

5.1 算法对比

算法	特点
线性回归	简单、可解释
岭回归	带 L2 正则
决策树回归	非线性
随机森林回归	稳健
GBT 回归	高精度

5.2 评估

指标	说明
RMSE	均方根误差
MAE	平均绝对误差
R2	拟合优度

---

六、聚类算法

6.1 算法对比

算法	特点	参数
KMeans	快、需定 K	k、seed
Bisecting KMeans	分层、稳定	k
GaussianMixture	概率模型	k
LDA	主题模型	topics

6.2 KMeans 使用

val kmeans = new KMeans().setK(5).setSeed(1L)
val model = kmeans.fit(features)
val cost = model.computeCost(features)  // 评估

6.3 评估

指标	说明
WSSSE	簇内平方和（越小越紧）
Silhouette	轮廓系数（越大越好）

---

七、协同过滤

7.1 原理

基于用户-物品评分矩阵，预测缺失评分：

矩阵分解：评分 ≈ 用户因子 × 物品因子

7.2 ALS 使用

import org.apache.spark.ml.recommendation.ALS

val als = new ALS()
  .setMaxIter(10).setRegParam(0.1)
  .setUserCol("userId").setItemCol("movieId").setRatingCol("rating")

val model = als.fit(train)
val predictions = model.transform(test)

参数	说明
rank	因子维度
regParam	正则化系数
implicitPrefs	是否隐式反馈
coldStartStrategy	冷启动处理

7.3 评估

指标	说明
RMSE	评分预测误差
Precision@K	推荐命中率

---

八、模型保存与加载

8.1 保存

model.write.overwrite().save("hdfs:///models/lr_model")

8.2 加载

val model = LogisticRegressionModel.load("hdfs:///models/lr_model")

注意	说明
Pipeline 保存	保存 PipelineModel 含全部阶段
版本兼容	模型与 Spark 版本相关
部署	加载后 transform 即可线上预测

---

九、训练流程最佳实践

1. 数据准备：清洗、类型转换
2. 特征工程：Indexer/OneHot/Assembler/Scaler
3. 划分训练集/测试集（8:2）
4. 构建 Pipeline
5. 交叉验证选参（CrossValidator）
6. 评估：分类 AUC / 回归 RMSE
7. 保存模型

9.1 交叉验证

val cv = new CrossValidator()
  .setEstimator(pipeline)
  .setEvaluator(evaluator)
  .setNumFolds(5)
  .setEstimatorParamMaps(paramGrid)

---

常见问题速查

问题	原因与处理
特征列不存在	检查 VectorAssembler 列名与类型
类别标签问题	用 StringIndexer 转索引
冷启动预测 NaN	设置 coldStartStrategy=drop
收敛慢	调 maxIter、正则化
数据倾斜	采样或调整分区
模型版本不兼容	同版本 Spark 训练与加载