机器学习库（MLlib）指南
MLlib是Spark的机器学习（ML）库。其目标是使实际的机器学习可扩展和容易。在高层次上，它提供了如下工具：
-
ML算法：通用学习算法，如分类，回归，聚类和协同过滤
-
特征提取，特征提取，转换，降维和选择
-
管道：用于构建，评估和调整ML管道的工具
-
持久性：保存和加载算法，模型和管道
-
实用程序：线性代数，统计，数据处理等
公告：基于DataFrame的API是主要的API
MLlib基于RDD的API现在处于维护模式。
从Spark 2.0开始，包中的基于RDD的API spark.mllib已进入维护模式。Spark的主要机器学习API现在是包中的基于DataFrame的API spark.ml。
有什么影响？
-
MLlib将仍然支持基于RDD的API
spark.mllib并修复错误。 -
MLlib不会将新功能添加到基于RDD的API。
-
在Spark 2.x版本中，MLlib将向基于DataFrame的API添加功能，以便与基于RDD的API达成功能对等。
-
达到功能对等（大致估计为Spark 2.2）后，基于RDD的API将被弃用。
-
基于RDD的API预计将在Spark 3.0中被删除。
为什么MLlib切换到基于DataFrame的API？
-
DataFrames提供比RDD更友好的API。DataFrame的许多优点包括Spark数据源，SQL / DataFrame查询，Tungsten和Catalyst优化以及跨语言的统一API。
-
MLlib的基于DataFrame的API提供跨ML算法和跨多种语言的统一API。
-
数据框便于实际的ML管线，特别是功能转换。
什么是“Spark ML”？
-
“Spark ML”不是一个正式的名字，偶尔用于指代基于MLlib DataFrame的API。这主要是由于
org.apache.spark.ml基于DataFrame的API所使用的Scala包名以及我们最初用来强调管道概念的“Spark ML Pipelines”术语。
MLlib是否被弃用？
-
编号MLlib包括基于RDD的API和基于DataFrame的API。基于RDD的API现在处于维护模式。
依赖
MLlib使用线性代数包Breeze，它依赖于 netlib-java进行优化的数值处理。如果本机库在运行时不可用，您将看到一条警告消息，而将使用纯JVM实现。
由于运行时专有二进制文件的授权问题，netlib-java默认情况下，我们不包含本地代理。要配置netlib-java/ Breeze以使用系统优化的二进制文件，请包括 com.github.fommil.netlib:all:1.1.2（或者构建Spark -Pnetlib-lgpl）作为项目的依赖项，并阅读netlib-java文档以获取平台的其他安装说明。
要在Python中使用MLlib，您将需要NumPy 1.4或更高版本。
推荐阅读：
4，Kafka源码系列之topic创建分区分配及leader选举
http://blog.csdn.net/rlnLo2pNEfx9c/article/details/78692975
以上就是ml和mllib的主要异同点。下面是ml和mllib逻辑回归的例子，可以对比看一下， 虽然都是模型训练和预测，但是画风很不一样。
sparse_data = [
LabeledPoint(0.0, SparseVector(2, {0: 0.0})),
LabeledPoint(1.0, SparseVector(2, {1: 1.0})),
LabeledPoint(0.0, SparseVector(2, {0: 1.0})),
LabeledPoint(1.0, SparseVector(2, {1: 2.0}))
]
lrm = LogisticRegressionWithSGD.train(sc.parallelize(sparse_data), iterations=10)
lrm.predict(array([0.0, 1.0]))
lrm.predict(array([1.0, 0.0]))
lrm.predict(SparseVector(2, {1: 1.0}))
lrm.predict(SparseVector(2, {0: 1.0}))
import os, tempfile
path = tempfile.mkdtemp()
lrm.save(sc, path)
sameModel = LogisticRegressionModel.load(sc, path)
sameModel.predict(array([0.0, 1.0]))
sameModel.predict(SparseVector(2, {0: 1.0}))
from shutil import rmtree
try:
rmtree(path)
except:
pass
multi_class_data = [
LabeledPoint(0.0, [0.0, 1.0, 0.0]),
LabeledPoint(1.0, [1.0, 0.0, 0.0]),
LabeledPoint(2.0, [0.0, 0.0, 1.0])
]
data = sc.parallelize(multi_class_data)
mcm = LogisticRegressionWithLBFGS.train(data, iterations=10, numClasses=3)
mcm.predict([0.0, 0.5, 0.0])
mcm.predict([0.8, 0.0, 0.0])
mcm.predict([0.0, 0.0, 0.3])
ml中的逻辑回归的例子
from pyspark.sql import Row
from pyspark.ml.linalg import Vectors
bdf = sc.parallelize([
Row(label=1.0, weight=2.0, features=Vectors.dense(1.0)),
Row(label=0.0, weight=2.0, features=Vectors.sparse(1, [], []))]).toDF()
blor = LogisticRegression(maxIter=5, regParam=0.01, weightCol="weight")
blorModel = blor.fit(bdf)
blorModel.coefficients
DenseVector([5.5 ])
blorModel.intercept
-2.68
mdf = sc.parallelize([
Row(label=1.0, weight=2.0, features=Vectors.dense(1.0)),
Row(label=0.0, weight=2.0, features=Vectors.sparse(1, [], [])),
Row(label=2.0, weight=2.0, features=Vectors.dense(3.0))]).toDF()
mlor = LogisticRegression(maxIter=5, regParam=0.01, weightCol="weight",
family="multinomial")
mlorModel = mlor.fit(mdf)
print(mlorModel.coefficientMatrix)
DenseMatrix([[-2.3 ],
[ 0.2 ],
[ 2.1 ]])
mlorModel.interceptVector
DenseVector([2.0 , 0.8 , -2.8 ])
test0 = sc.parallelize([Row(features=Vectors.dense(-1.0))]).toDF()
result = blorModel.transform(test0).head()
result.prediction
0.0
result.probability
DenseVector([0.99 , 0.00 ])
result.rawPrediction
DenseVector([8.22 , -8.22 ])
test1 = sc.parallelize([Row(features=Vectors.sparse(1, [0], [1.0]))]).toDF()
blorModel.transform(test1).head().prediction
1.0
blor.setParams("vector")
Traceback (most recent call last):
TypeError: Method setParams forces keyword arguments.
lr_path = temp_path + "/lr"
blor.save(lr_path)
lr2 = LogisticRegression.load(lr_path)
lr2.getMaxIter()
model_path = temp_path + "/lr_model"
blorModel.save(model_path)
model2 = LogisticRegressionModel.load(model_path)
blorModel.coefficients[0] == model2.coefficients[0]
True
blorModel.intercept == model2.intercept
True