Spark 0.9.1 MLLib 机器学习库简介 - 推酷
Spark 0.9.1 MLLib 机器学习库简介
这篇Spark MLLib 机器学习库的简介翻译自
。感谢 @明风 的悉心的校对修改。
Spark 0.9.1 MLLib
机器学习库简介
用Scala调用MLLib
用Python调用MLLib
MLlib 是Spark对常用的机器学习算法的实现库，同时包括相关的测试和数据生成器。MLlib 目前支持四种常见的机器学习问题：二元分类，回归，聚类以及协同过滤，同时也包括一个底层的梯度下降优化基础算法。本指南将会简要介绍 MLlib 中所支持的功能，并给出相应的调用 MLlib 的例子。
MLlib 将会调用
线性代数库，这个库本身依赖于原生的 Fortran 程序。如果你的节点中没有这些库，你也许会需要安装
。如果程序没有办法自动检测到这些库，MLlib 将会抛出链接错误的异常。
如果想用 Python 调用 MLlib,你需要安装
1.7 或者更新的版本。
二元分类是一个监督学习问题。在这个问题中，我们希望将实体归类到两个独立的类别或标签的其中一个中，例如判断一个邮件是否是垃圾邮件。这个问题涉及在一组被
的样例运行一个学习
，例如一组由(数字)特征和(相关的)类别标签所代表的实体。这个算法将会返回一个训练好的模型，该模型能够对标签未知的新个体进行潜在标签预测。
MLlib 目前支持两个适用于二元分类的标准模型家族：
，同时也包括分别适用与这两个模型家族的
变体。这些训练算法都利用了一个底层的梯度下降基础算法（描述如下）。二元分类算法的输入值是一个正则项参数(regParam) 和多个与梯度下降相关的参数(
stepSize, numIterations, miniBatchFraction
目前可用的二元分类算法：
线性回归是另一个经典的监督学习问题。在这个问题中，每个个体都有一个与之相关联的实数标签（而在二元分类中个体的标签都是二元的），并且我们希望在给出用于表示这些实体的数值特征后，所预测出的标签值可以尽可能接近实际值。MLlib支持线性回归和与之相关的 L1 (
) 正则化的变体。MLlib中的回归算法也利用了底层的梯度下降基础算法（描述如下），输入参数与上述二元分类算法一致。
目前可用的线性回归算法：
聚类是一个非监督学习问题，在这个问题上，我们的目标是将一部分实体根据某种意义上的相似度和另一部分实体聚在一起。聚类通常被用于探索性的分析，或者作为层次化监督学习管道网(hierarchical supervised learning pipeline) 的一个组件（其中每一个类簇都会用与训练不同的分类器或者回归模型)。 MLlib 目前已经支持作为最被广泛使用的聚类算法之一的
聚类算法，根据事先定义的类簇个数，这个算法能对数据进行聚类。MLlib 的实现中包含一个
方法的并行化变体
。 MLlib 里面的实现有如下的参数：
是所需的类簇的个数。
maxIterations
是最大的迭代次数。
initializationMode
这个参数决定了是用随机初始化还是通过 k-means|| 进行初始化。
是跑 k-means 算法的次数（k-mean 算法不能保证能找出最优解，如果在给定的数据集上运行多次，算法将会返回最佳的结果）。
initializiationSteps
决定了 k-means|| 算法的步数。
决定了判断 k-means 是否收敛的距离阀值。
目前可用的聚类算法：
常被应用于推荐系统。这些技术旨在补充用户-商品关联矩阵中所缺失的部分。MLlib当前支持基于模型的协同过滤，其中用户和商品通过一小组隐语义因子进行表达，并且这些因子也用于预测缺失的元素。为此，我们实现了
来学习这些隐性语义因子。在 MLlib 中的实现有如下的参数:
是用于并行化计算的分块个数 (设置为-1为自动配置)。
是模型中隐语义因子的个数。
iterations
是迭代的次数。
是ALS的正则化参数。
implicitPrefs
决定了是用
ALS的版本还是用适用
数据集的版本。
是一个针对于
ALS 版本的参数，这个参数决定了偏好行为强度的
隐性反馈 vs 显性反馈
基于矩阵分解的协同过滤的标准方法一般将用户商品矩阵中的元素作为用户对商品的显性偏好。
在许多的现实生活中的很多场景中，我们常常只能接触到
隐性的反馈
（例如游览，点击，购买，喜欢，分享等等）在 MLlib 中所用到的处理这种数据的方法来源于文献：
。 本质上，这个方法将数据作为二元偏好值和
的一个结合，而不是对评分矩阵直接进行建模。因此，评价就不是与用户对商品的显性评分而是和所观察到的用户偏好强度关联了起来。然后，这个模型将尝试找到隐语义因子来预估一个用户对一个商品的偏好。
目前可用的协同过滤的算法：
梯度下降基础算法
（及其随机的变种）是非常适用于大型分布式计算的一阶优化方案。梯度下降旨在通过向一个函数当前点（当前的参数值）的负梯度方向移动的方式迭代地找到这个函数的本地最优解。MLlib 以梯度下降作为一个底层的基础算法，在上面开发了各种机器学习算法。梯度下降算法有如下的参数：
这个类是用来计算要被优化的函数的随机梯度(如：相对于单一训练样本当前的参数值)。MLlib 包含常见损失函数 (hinge, logistic, least-squares) 的梯度类。梯度类将训练样本，其标签，以及当前的参数值作为输入值。
是在梯度下降的每一次迭代中更新权重的类。MLlib 包含适用于无正则项，L1 正则项和 L2 正则项3种情况下的类。
是一个表示梯度下降初始步长的数值。MLlib 中所有的更新器第 t 步的步长等于 stepSize / sqrt(t)。
numIterations
表示迭代的次数。
是在使用L1，L2 正则项时的正则化参数。
miniBatchFraction
是每一次迭代中用来计算梯度的数据百分比。
目前可用的梯度下降算法：
用Scala调用MLLib
下面的代码段可以在
spark-shell
下面的代码段演示了如何导入一份样本数据集，使用算法对象中的静态方法在训练集上执行训练算法，在所得的模型上进行预测并计算训练误差。
import org.apache.spark.SparkContext
import org.apache.spark.mllib.classification.SVMWithSGD
import org.apache.spark.mllib.regression.LabeledPoint
// Load and parse the data file
val data = sc.textFile(&mllib/data/sample_svm_data.txt&)
val parsedData = data.map { line =&
val parts = line.split(' ')
LabeledPoint(parts(0).toDouble, parts.tail.map(x =& x.toDouble).toArray)
// Run training algorithm to build the model
val numIterations = 20
val model = SVMWithSGD.train(parsedData, numIterations)
// Evaluate model on training examples and compute training error
val labelAndPreds = parsedData.map { point =&
val prediction = model.predict(point.features)
(point.label, prediction)
val trainErr = labelAndPreds.filter(r =& r._1 != r._2).count.toDouble / parsedData.count
println(&Training Error = & + trainErr)
SparkContext
classification
SVMWithSGD
regression
LabeledPoint
// Load and parse the data file
&mllib/data/sample_svm_data.txt&
parsedData
LabeledPoint
// Run training algorithm to build the model
numIterations
SVMWithSGD
parsedData
numIterations
// Evaluate model on training examples and compute training error
labelAndPreds
parsedData
prediction
prediction
labelAndPreds
parsedData
&Training Error = &
默认情况下，这个
SVMWithSGD.train()
方法使用正则参数为 1.0 的 L2 正则项。如果我们想配置这个算法，我们可以通过直接新建一个新的对象，并调用setter的方法，进一步个性化设置
SVMWithSGD
。所有其他的 MLlib 算法也是通过这样的方法来支持个性化的设置。比如，下面的代码给出了一个正则参数为0.1的 L1 正则化SVM变体，并且让这个训练算法迭代200遍。
import org.apache.spark.mllib.optimization.L1Updater
val svmAlg = new SVMWithSGD()
svmAlg.optimizer.setNumIterations(200)
.setRegParam(0.1)
.setUpdater(new L1Updater)
val modelL1 = svmAlg.run(parsedData)
optimization
SVMWithSGD
setNumIterations
setRegParam
setUpdater
parsedData
下面这个例子演示了如何导入训练集数据，将其解析为带标签点的RDD。然后，使用
LinearRegressionWithSGD
算法来建立一个简单的线性模型来预测标签的值。最后我们计算了均方差来评估预测值与实际值的
import org.apache.spark.mllib.regression.LinearRegressionWithSGD
import org.apache.spark.mllib.regression.LabeledPoint
// Load and parse the data
val data = sc.textFile(&mllib/data/ridge-data/lpsa.data&)
val parsedData = data.map { line =&
val parts = line.split(',')
LabeledPoint(parts(0).toDouble, parts(1).split(' ').map(x =& x.toDouble).toArray)
// Building the model
val numIterations = 20
val model = LinearRegressionWithSGD.train(parsedData, numIterations)
// Evaluate model on training examples and compute training error
val valuesAndPreds = parsedData.map { point =&
val prediction = model.predict(point.features)
(point.label, prediction)
val MSE = valuesAndPreds.map{ case(v, p) =& math.pow((v - p), 2)}.reduce(_ + _)/valuesAndPreds.count
println(&training Mean Squared Error = & + MSE)
regression
LinearRegressionWithSGD
regression
LabeledPoint
// Load and parse the data
&mllib/data/ridge-data/lpsa.data&
parsedData
LabeledPoint
// Building the model
numIterations
LinearRegressionWithSGD
parsedData
numIterations
// Evaluate model on training examples and compute training error
valuesAndPreds
parsedData
prediction
prediction
valuesAndPreds
valuesAndPreds
&training Mean Squared Error = &
类似的，你也可以使用
RidgeRegressionWithSGD
LassoWithSGD
这两个算法，并比较这些算法在训练集上的均方差。
在下面的例子中，在载入和解析数据之后，我们使用 KMeans 对象来将数据聚类到两个类簇当中。所需的类簇个数会被传递到算法中。然后我们将计算集内均方差总和 (WSSSE). 你可以通过增加类簇的个数
来减小误差。 实际上，最优的类簇数通常是 1，因为这一点通常是WSSSE图中的 “低谷点”。
import org.apache.spark.mllib.clustering.KMeans
// Load and parse the data
val data = sc.textFile(&kmeans_data.txt&)
val parsedData = data.map( _.split(' ').map(_.toDouble))
// Cluster the data into two classes using KMeans
val numIterations = 20
val numClusters = 2
val clusters = KMeans.train(parsedData, numClusters, numIterations)
// Evaluate clustering by computing Within Set Sum of Squared Errors
val WSSSE = puteCost(parsedData)
println(&Within Set Sum of Squared Errors = & + WSSSE)
clustering
// Load and parse the data
&kmeans_data.txt&
parsedData
// Cluster the data into two classes using KMeans
numIterations
numClusters
parsedData
numClusters
numIterations
// Evaluate clustering by computing Within Set Sum of Squared Errors
computeCost
parsedData
&Within Set Sum of Squared Errors = &
在下面的例子中，我们导入的训练集中，数据每一行由一个用户，一个商品和相应的评分组成。假设评分是显性的，在这种情况下我们使用默认的
ALS.train()
方法。我们通过计算预测出的评分的均方差来评估这个推荐模型。
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating
// Load and parse the data
val data = sc.textFile(&mllib/data/als/test.data&)
val ratings = data.map(_.split(',') match {
case Array(user, item, rate) =& Rating(user.toInt, item.toInt, rate.toDouble)
// Build the recommendation model using ALS
val numIterations = 20
val model = ALS.train(ratings, 1, 20, 0.01)
// Evaluate the model on rating data
val usersProducts = ratings.map{ case Rating(user, product, rate) =& (user, product)}
val predictions = model.predict(usersProducts).map{
case Rating(user, product, rate) =& ((user, product), rate)
val ratesAndPreds = ratings.map{
case Rating(user, product, rate) =& ((user, product), rate)
}.join(predictions)
val MSE = ratesAndPreds.map{
case ((user, product), (r1, r2)) =& math.pow((r1- r2), 2)
}.reduce(_ + _)/ratesAndPreds.count
println(&Mean Squared Error = & + MSE)
recommendation
recommendation
// Load and parse the data
&mllib/data/als/test.data&
// Build the recommendation model using ALS
numIterations
// Evaluate the model on rating data
usersProducts
predictions
usersProducts
ratesAndPreds
predictions
ratesAndPreds
ratesAndPreds
&Mean Squared Error = &
如果这个评分矩阵是通过其他的信息来源（如从其他的信号中提取出来的）所获得，你也可以使用
trainImplicit
的方法来得到更好的结果。
val model = ALS.trainImplicit(ratings, 1, 20, 0.01)
trainImplicit
用Java调用MLLib
所有 MLlib 中的算法都是对Java友好的，因此你可以用在 Scala 中一样的方法来导入和调用这些算法。唯一要注意的是，这些算法的输入值是
对象，而在 Spark Java API 中用了分离的
类。你可以在你的 JavaRDD对象中调用
的方法来将
用Python调用MLLib
下面的列子可以在
PySpark shell
中得到测试。
下面的代码段表明了如何导入一份样本数据集，使用算法对象中的静态方法在训练集上执行训练算法，在所得的模型上进行预测并计算训练误差。
from pyspark.mllib.classification import LogisticRegressionWithSGD
from numpy import array
# Load and parse the data
data = sc.textFile(&mllib/data/sample_svm_data.txt&)
parsedData = data.map(lambda line: array([float(x) for x in line.split(' ')]))
model = LogisticRegressionWithSGD.train(parsedData)
# Build the model
labelsAndPreds = parsedData.map(lambda point: (int(point.item(0)),
model.predict(point.take(range(1, point.size)))))
# Evaluating the model on training data
trainErr = labelsAndPreds.filter(lambda (v, p): v != p).count() / float(parsedData.count())
print(&Training Error = & + str(trainErr))
classification
LogisticRegressionWithSGD
# Load and parse the data
&mllib/data/sample_svm_data.txt&
parsedData
LogisticRegressionWithSGD
parsedData
# Build the model
labelsAndPreds
parsedData
# Evaluating the model on training data
labelsAndPreds
parsedData
&Training Error = &
下面这个例子给出了如何导入训练集数据，将其解析为带标签点的RDD。然后，这个例子使用了
LinearRegressionWithSGD
算法来建立一个简单的线性模型来预测标签的值。我们在最后计算了均方差来评估预测值与实际值的
from pyspark.mllib.regression import LinearRegressionWithSGD
from numpy import array
# Load and parse the data
data = sc.textFile(&mllib/data/ridge-data/lpsa.data&)
parsedData = data.map(lambda line: array([float(x) for x in line.replace(',', ' ').split(' ')]))
# Build the model
model = LinearRegressionWithSGD.train(parsedData)
# Evaluate the model on training data
valuesAndPreds = parsedData.map(lambda point: (point.item(0),
model.predict(point.take(range(1, point.size)))))
MSE = valuesAndPreds.map(lambda (v, p): (v - p)**2).reduce(lambda x, y: x + y)/valuesAndPreds.count()
print(&Mean Squared Error = & + str(MSE))
regression
LinearRegressionWithSGD
# Load and parse the data
&mllib/data/ridge-data/lpsa.data&
parsedData
# Build the model
LinearRegressionWithSGD
parsedData
# Evaluate the model on training data
valuesAndPreds
parsedData
valuesAndPreds
valuesAndPreds
&Mean Squared Error = &
类似的，你也可以使用
RidgeRegressionWithSGD
LassoWithSGD
这两个算法，并比较这些算法在训练集上的均方差。
在下面的例子中，在载入和解析数据之后，我们使用 KMeans对象来将数据聚类到两个类簇当中。所需的类簇个数被传递到算法中。然后我们将计算集内均方差总和(WSSSE). 你可以通过增加类簇的个数
来减小误差。 实际上，最优的类簇数通常是 1，因为这一点通常是WSSSE图中的”低谷点”。
from pyspark.mllib.clustering import KMeans
from numpy import array
from math import sqrt
# Load and parse the data
data = sc.textFile(&kmeans_data.txt&)
parsedData = data.map(lambda line: array([float(x) for x in line.split(' ')]))
# Build the model (cluster the data)
clusters = KMeans.train(parsedData, 2, maxIterations=10,
runs=30, initialization_mode=&random&)
# Evaluate clustering by computing Within Set Sum of Squared Errors
def error(point):
center = clusters.centers[clusters.predict(point)]
return sqrt(sum([x**2 for x in (point - center)]))
WSSSE = parsedData.map(lambda point: error(point)).reduce(lambda x, y: x + y)
print(&Within Set Sum of Squared Error = & + str(WSSSE))
clustering
# Load and parse the data
&kmeans_data.txt&
parsedData
# Build the model (cluster the data)
parsedData
maxIterations
initialization_mode
# Evaluate clustering by computing Within Set Sum of Squared Errors
parsedData
&Within Set Sum of Squared Error = &
在下面的例子中，我们导入的训练集中，数据每一行由一个用户，一个商品和相应的评分组成。假设评分是显性的，在这种情况下我们使用默认的&
ALS.train()
方法。我们通过计算预测出的评分的均方差来评估这个推荐模型。
from pyspark.mllib.recommendation import ALS
from numpy import array
# Load and parse the data
data = sc.textFile(&mllib/data/als/test.data&)
ratings = data.map(lambda line: array([float(x) for x in line.split(',')]))
# Build the recommendation model using Alternating Least Squares
model = ALS.train(ratings, 1, 20)
# Evaluate the model on training data
testdata = ratings.map(lambda p: (int(p[0]), int(p[1])))
predictions = model.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))
ratesAndPreds = ratings.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).reduce(lambda x, y: x + y)/ratesAndPreds.count()
print(&Mean Squared Error = & + str(MSE))
recommendation
# Load and parse the data
&mllib/data/als/test.data&
# Build the recommendation model using Alternating Least Squares
# Evaluate the model on training data
predictions
predictAll
ratesAndPreds
predictions
ratesAndPreds
ratesAndPreds
&Mean Squared Error = &
如果这个评分矩阵是通过其他的信息来源（如从其他的信号中提取出来的）所获得，你也可以使用
trainImplicit
的方法来得到更好的结果。
# Build the recommendation model using Alternating Least Squares based on implicit ratings
model = ALS.trainImplicit(ratings, 1, 20)
# Build the recommendation model using Alternating Least Squares based on implicit ratings
trainImplicit
已发表评论数()
请填写推刊名
描述不能大于100个字符!
权限设置： 公开
仅自己可见
正文不准确
标题不准确
排版有问题
主题不准确
没有分页内容
图片无法显示
视频无法显示
与原文不一致spark 对设备有最低要求么？
standalone模式下在提交spark examples中pi.py时：
:/usr/lib/spark#./bin/spark-submit --master spark://master:7077 --executor-memory 512m examples/src/main/python/pi.py就会出现如下的问题
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x728, 0) error='Cannot allocate memory' (errno=12)
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map
bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/lib/spark/hs_err_pid16414.log
我怎么修改spark.executor.memory的大小都没有效果，我的配置是这样的三台机子都是1g内存，是因为配置不够的原因吗？求大神速速解答啊
已有帐号？
无法登录？
社交帐号登录主题信息（必填）
主题描述（最多限制在50个字符）
申请人信息（必填）
申请信息已提交审核，请注意查收邮件，我们会尽快给您反馈。
如有疑问，请联系
傻丫头和高科技产物小心翼翼的初恋
如今的编程是一场程序员和上帝的竞赛，程序员要开发出更大更好、傻瓜都会用到软件。而上帝在努力创造出更大更傻的傻瓜。目前为止，上帝是赢的。个人网站：。个人QQ群：、
CSDN &《程序员》编辑/记者，我的邮箱
一只文艺范的软件攻城狮，Keep Learn，Always.
个人大数据技术博客：《Spark上机器学习方法的模式与性能》Intel尹绪森，6月18日19:30-21:30 - 小象学院 - 中国最专业的Hadoop,Spark大数据在线教育平台——权威课程：Hadoop培训,Spark培训,HBase培训,Hive培训,Mahout培训等 - Powered By EduSoho
主题：Spark上机器学习方法的模式与性能
时间：日（周三）19:30-21:30
课程介绍：
处理大数据, 并从数据中获得知识是大数据时代的两大基石. 源于加州大学AMPLab的Spark无疑是大数据处理平台的新锐代表. 价值千万美金的Spark项目自诞生之日就备受瞩目, 甚至老东家AMPLab也顺势成立初创公司Databricks, 希望为Spark持续发展保驾护航. 随着Spark 1.0的发布, Spark已经在批处理,流计算, 机器学习, 图计算, SQL查询等一系列领域大展拳脚, 并随着愈发活跃的开发者社区以及Twitter, Adobe, Intel, Amazon, Redhat等公司的加入而渐成气候.
在大数据世界的另一端, 机器学习, 尤其是深度学习的极大发展让知识获取如沐春风. 比之于Hadoop, Spark对machine Learning也有更好的支持. 构建在Spark上的MLlib是Spark社区中机器学习方面的代表.从模型上来看, 广义线性模型, 决策树, 矩阵分解全面覆盖; 从数值优化上来看, 梯度下降, 牛顿方法, ADMM等一样不缺.MLlib逐渐走向成熟, 接口稳定, 算法表现优异. 在业界也饱受关注的目光.
机器学习算法有两面, 一面模型一面优化. 模型分为结构和参数, 结构有如深度学习中的卷积神经网络, 抑或浅层学习中的线性模型. 参数也有向量式的参数, 以及矩阵式的参数. 结构和参数交织, 再加上分布式平台Spark上的RDD表达的艺术, 织筑了分布式机器学习的画卷. 而言及优化, MIT教授, 第四范式的合伙人和CTO, 数据库领域宗师Mike Stonebraker在一次关于SciDB的访谈时说过, “Whether you are doing regression, singular value decomposition, finding eigenvectors, or doing operations on graphs, you are performing a sequence of matrix operations.” 一切关于计算的活动最终还是回归到矩阵计算本身. 因此可以说, 同样的算法下, 很大程度上是分布式访存模式影响了最终的性能. 而访存模式来源于模型与优化.
此次就Spark上机器学习方法的模式与性能为题, 讲述Spark上机器学习实现的一般方法并总结访存模式,希望为Spark上机器学习算法的开发带来一定的帮助. 内容要点包含Spark MLlib 1.0 简介, 常见机器学习算法的结构分析, MLlib访存模式等深入介绍等.
Spark MLlib 1.0简介
向量库现状, 使用, 及其分析, 性能分析
模型表达, tree, GLM, LDA, kmeans, NaiveBayes
优化方法: SGD, L-BFGS
Evaluation: AUC
常见机器学习算法结构分析
广义线性模型
深度学习模型
MLlib访存模式分析
广义线性模型的访存模式
ALS模型的访存模式
LDA模型的访存模式
尹绪森，Intel中国研究院 工程师
熟悉并热爱机器学习相关内容，对自然语言处理、推荐系统等有所涉猎。
目前致力于机器学习算法并行、凸优化层面的算法优化问题，以及大数据平台性能调优。
对Spark、Mahout、GraphLab等开源项目有所尝试和理解，并希望从优化层向下，系统层向上对并行算法及平台做出贡献。
报名方式：
选择“购买课程”，价格为0。
联系我们：
微信公众号：ChinaHadoop
学完了课时
收藏了课程
开始学习课时
开始学习课时