转载请注明出处：http://blog.csdn.net/gamer_gyt
博主微博：http://weibo.com/234654758
Github：https://github.com/thinkgamer
公众号：搜索与推荐Wiki

个人网站：http://thinkgamer.github.io

【Spark排序算法系列】主要介绍的是目前推荐系统或者广告点击方面用的比较广的几种算法，和他们在Spark中的应用实现，本篇文章主要介绍LR算法。

本系列还包括（持续更新）：

Spark排序算法系列之GBDT（梯度提升决策树）
Spark排序算法系列之模型融合（GBDT+LR）
Spark排序算法系列之XGBoost
Spark排序算法系列之FTRL（Follow-the-regularized-Leader）
Spark排序算法系列之FM与FFM

背景

逻辑回归（Logistic Regression，LR）是较早应用在推荐排序上的，其属于线性模型，模型简单，可以引入海量离散特征，这样的好处就是模型可以考虑更加细节或者说针对具体个体的因素。如果想要引入非线性因素需要做特征交叉，这样很容易产生百亿特征，在很早之前ctr就主要靠堆人力搞特征工程工作来持续优化效果。

虽然目前在工业界LR应用的并不多，但是对于初学者,一些中小企业或者应用场景不需要负责排序模型的时候，LR扔不失为一个不错的选择。

关于LR的算法原理，这里不做过多说明，可参考：

回归分析之逻辑回归-Logistic Regression
线性模型篇之Logistic Regression数学公式推导

LR介绍

LR的数学表达式可以简写为：

L(w,x,y)=log(1+exp(−ywTx))L(w,x,y)=log(1+exp(-yw^Tx)) L(w,x,y)=log(1+exp(−ywTx))
对于二分类模型，LR是一个分类算法，模型计算得到预测值后会通过以下函数进转化。
f(z)=11+e−zxf(z) = \frac{1}{1+e^{-zx}} f(z)=1+e−zx1
如果L(w,x,y) > 0.5 则是1 否则为0。当然在实际应用过程中，并不是一定取0.5作为界限值，而是根据实际情况进行调整。

二进制回归可以转化为多分类回归问题。关于多分类介绍和基于Spark实现多分类可参考多分类实现方式介绍和在Spark上实现多分类逻辑回归（Multinomial Logistic Regression）

在Spark.mllib包中提供了两种LR分类模型，分别是：

mini-batch gradient descent（LogisticRegressionWithLBFGS）
L-BFGS（LogisticRegressionWithSGD）

但官方给出的建议是：推荐使用LBFGS，因为基于LBFGS的LR比基于SGD的能更快的收敛。其原话如下：

We implemented two algorithms to solve logistic regression: mini-batch gradient descent and L-BFGS. We recommend L-BFGS over mini-batch gradient descent for faster convergence.

而且LRWithLBFGS不仅支持二分类还支持多分类，但LRWithSGD只支持二分类。所以后续只介绍下Spark mllib中的LogisticRegressionWithLBFGS相关操作。

mllib中的LRWithLBFGS

设置变量和创建spark对象

val file = "data/sample_libsvm_data.txt"
val model_path = "model/lr/"
val model_param = "numInterations:5,regParam:0.1,updater:SquaredL2Updater,gradient:LogisticGradient"val spark = SparkSession.builder().master("local[5]").appName("LogisticRegression_Model_Train").getOrCreate()
Logger.getRootLogger.setLevel(Level.WARN)

拆分数据集

// 记载数据集 并拆分成训练集和测试集
val data = MLUtils.loadLibSVMFile(spark.sparkContext,file).randomSplit(Array(0.7,0.3))
val (train, test) = (data(0), data(1))

LRWithLBFGS模型设置参数

// 定义分类的数目，默认为2，是logisticregression的参数
private var numClass: Int = 2
// 定义是否添加截距,默认值为false，是logisticregression的参数
private var isAddIntercept: Option[Boolean] = None
// 定义是否在训练模型前进行验证，是logisticregression的参数
private var isValidateData: Option[Boolean] = None// 定义迭代的次数，默认值是100，LBFGS的参数
private var numInterations: Option[Int] = None
// 定义正则化系数值，默认值是0.0，LBFGS的参数
private var regParam: Option[Double] = None
// 定义正则化参数，支持：L1Updater[L1]、SquaredL2Updater[L2]、SimpleUpdater[没有正则项]，LBFGS的参数
private var updater: Option[String] = None
// 定义计算梯度的方式，支持：LogisticGradient、LeastSquaresGradient、HingeGradient ，LBFGS的参数
private var gradient: Option[String] = None
// 人工定义的收敛阈值
private var threshold:Option[Double]=None
// 定义模型收敛阈值，默认为 10^-6
private var convergenceTol: Double= 1.0e-6

创建模型

def createLRModel(model_param: String): LogisticRegressionWithLBFGS={// 设置模型参数val optimizer = new LROptimizer()optimizer.parseString(model_param)println(s"模型训练参数为：${optimizer.toString}")// 创建模型并指定相关参数val LRModel = new LogisticRegressionWithLBFGS()// 设置分类数目LRModel.setNumClasses(optimizer.getNumClass)// 设置是否添加截距if(optimizer.getIsAddIntercept.nonEmpty) {LRModel.setIntercept(optimizer.getIsAddIntercept.get)}// 设置是否进行验证模型if(optimizer.getIsValidateData.nonEmpty){LRModel.setValidateData(optimizer.getIsValidateData.get)}// 设置迭代次数if(optimizer.getNumInterations.nonEmpty){LRModel.optimizer.setNumIterations((optimizer.getNumInterations.get))}// 设置正则项参数if(optimizer.getRegParam.nonEmpty) { LRModel.optimizer.setRegParam(optimizer.getRegParam.get) }// 设置正则化参数if(optimizer.getUpdater.nonEmpty){optimizer.getUpdater match {case Some("L1Updater") => LRModel.optimizer.setUpdater( new L1Updater())case Some("SquaredL2Updater") => LRModel.optimizer.setUpdater(new SquaredL2Updater())case Some("SimpleUpdater") => LRModel.optimizer.setUpdater(new SimpleUpdater())case _ => LRModel.optimizer.setUpdater(new SquaredL2Updater())}}// 设置梯度计算方式if(optimizer.getGradient.nonEmpty){optimizer.getGradient match {case Some("LogisticGradient") => LRModel.optimizer.setGradient(new LogisticGradient())case Some("LeastSquaresGradient") => LRModel.optimizer.setGradient(new LeastSquaresGradient())case Some("HingeGradient") => LRModel.optimizer.setGradient(new HingeGradient())case _ => LRModel.optimizer.setGradient(new LogisticGradient())}}// 设置收敛阈值if(optimizer.getThreshold.nonEmpty){ LRModel.optimizer.setConvergenceTol(optimizer.getThreshold.get)}else {LRModel.optimizer.setConvergenceTol(optimizer.getConvergenceTol)}LRModel
}

模型效果评估

 def evaluteResult(result: RDD[(Double,Double,Double)]) :Unit = {// MSEval testMSE = result.map{ case(real, pre, _) => math.pow((real - pre), 2)}.mean()println(s"Test Mean Squared Error = $testMSE")// AUCval metrics = new BinaryClassificationMetrics(result.map(x => (x._2,x._1)).sortByKey(ascending = true),numBins = 2)println(s"0-1 label AUC is = ${metrics.areaUnderROC}")val metrics1 = new BinaryClassificationMetrics(result.map(x => (x._3,x._1)).sortByKey(ascending = true),numBins = 2)println(s"score-label AUC is = ${metrics1.areaUnderROC}")// 错误率val error = result.filter(x => x._1!=x._2).count().toDouble / result.count()println(s"error is = $error")// 准确率val accuracy = result.filter(x => x._1==x._2).count().toDouble / result.count()println(s"accuracy is = $accuracy")}

保存模型

 def saveModel(model: LogisticRegressionModel, model_path: String): Unit = {// 保存模型文件 objval out_obj = new ObjectOutputStream(new FileOutputStream(model_path+"model.obj"))out_obj.writeObject(model)// 保存模型信息val model_info=new BufferedWriter(new FileWriter(model_path+"model_info.txt"))model_info.write(model.toString())model_info.flush()model_info.close()// 保存模型权重val model_weights=new BufferedWriter(new FileWriter(model_path+"model_weights.txt"))model_weights.write(model.weights.toString)model_weights.flush()model_weights.close()println(s"模型信息写入文件完成，路径为：$model_path")}

加载模型

 def loadModel(model_path: String): Option[LogisticRegressionModel] = {try{val in = new ObjectInputStream( new FileInputStream(model_path) )val model = Option( in.readObject().asInstanceOf[LogisticRegressionModel] )in.close()println("Model Load Success")model}catch {case ex: ClassNotFoundException => {println(ex.printStackTrace())None}case ex: IOException => {println(ex.printStackTrace())println(ex)None}case _: Throwable => throw new Exception}}

使用加载的模型进行分值计算

 // 加载obj文件进行预测val model_new = loadModel(s"$model_path/model.obj")// 使用加载的模型进行样例预测val result_new = test.map(line =>{val pre_label = model_new.get.predict(line.features)// blas.ddot(x.length, x,1,y,1) (向量x的长度，向量x，向量x的索引递增间隔，向量y，向量y的索引递增间隔)val pre_score = blas.ddot(model.numFeatures, line.features.toArray, 1, model.weights.toArray, 1)val score = Math.pow(1+Math.pow(Math.E, -2 * pre_score), -1)(line.label, pre_label,score)} )result_new.take(2).foreach(println)

ml中的二分类LR

ml包中的LR既可以用来做二分类，也可以用来做多分类。

二分类对应：Binomial logistic regression
多分类对应：multinomial logistic regression

其中二分类可以通过Binomial logistic regression 和 multinomial logistic regression实现。

基于Binomial logistic regression的LR实现：

def BinaryModel(train: Dataset[Row], model_path: String, spark: SparkSession) = {// 创建模型val LRModel = new LogisticRegression().setMaxIter(20).setRegParam(0.3).setElasticNetParam(0.8)// 训练评估模型val model = LRModel.fit(train)evalute(model, train, spark)
}def evalute(model: LogisticRegressionModel, train: Dataset[Row], spark: SparkSession):Unit = {// 打印模型参数println(s"模型参数信息如下：\n ${model.parent.explainParams()} \n")println(s"Coefficients（系数）: ${model.coefficients}")println(s"Intercept（截距）: ${model.intercept}")// 查看训练集的预测结果 rawPrediction：row 计算的分值，probability：经过sigmoid转换后的概率val result = model.evaluate(train)result.predictions.show(10)// 将 label，0 值概率，predict label提取出来result.predictions.select("label","probability","prediction").rdd.map(row => (row.getDouble(0),row.get(1).asInstanceOf[DenseVector].toArray(0),row.getDouble(2))).take(10).foreach(println)// 模型评估val trainSummary = model.summaryval objectiveHistory = trainSummary.objectiveHistoryprintln("objectiveHistoryLoss:")objectiveHistory.foreach(loss => println(loss))val binarySummary = trainSummary.asInstanceOf[BinaryLogisticRegressionSummary]val roc = binarySummary.rocroc.show()println(s"areaUnderROC: ${binarySummary.areaUnderROC}")// Set the model threshold to maximize F-Measureval fMeasure = binarySummary.fMeasureByThresholdfMeasure.show(10)val maxFMeasure = fMeasure.select(max("F-Measure")).head().getDouble(0)import spark.implicits ._val bestThreshold = fMeasure.where($"F-Measure"===maxFMeasure).select("threshold").head().getDouble(0)model.setThreshold(bestThreshold)}

基于Multimial logistic regression的LR实现：

def BinaryModelWithMulti(train: Dataset[Row], model_path: String, spark: SparkSession) = {// 创建模型val LRModel = new LogisticRegression().setMaxIter(10).setRegParam(0.3).setElasticNetParam(0.8).setFamily("multinomial")// 训练模型val model = LRModel.fit(train)// 打印模型参数println(s"模型参数信息如下：\n ${model.parent.explainParams()} \n")println(s"Coefficients（系数）: ${model.coefficientMatrix}")println(s"Intercept（截距）: ${model.interceptVector}")
}

ml中的多分类LR

某条样本属于类别k的概率计算为：
P(Y=k∣X,βk,β0k)=eβk⋅X+β0k∑kj=0K−1eβkj⋅X+β0kjP(Y=k | X,\beta_k,\beta_{0k} ) = \frac { e^{ \beta _k \cdot X + \beta_{0k} }} { \sum_{k^j=0}^{K-1} e^{ \beta _{k^j} \cdot X + \beta_{0k^j}} } P(Y=k∣X,βk,β0k)=∑kj=0K−1eβkj⋅X+β0kjeβk⋅X+β0k

其中K表示类别，J表示特征个数

权重最小化使用的是最大似然函数，其更新公式如下：
minβ,β0−[∑i=1Lwi⋅logP(Y=yi∣Xi)]+λ[12(1−α)∣∣β∣∣22+α∣∣β∣∣1]\underset{ \beta , \beta_0 }{min} - [\sum_{i=1}^{L}w_i \cdot logP(Y=y_i|X_i) ] + \lambda [ \frac{1}{2}(1-\alpha )||\beta||_2^2 + \alpha ||\beta||_1] β,β0min−[i=1∑Lwi⋅logP(Y=yi∣Xi)]+λ[21(1−α)∣∣β∣∣22+α∣∣β∣∣1]

使用的数据集形式为：

1 1:-0.222222 2:0.5 3:-0.762712 4:-0.833333
1 1:-0.555556 2:0.25 3:-0.864407 4:-0.916667
1 1:-0.722222 2:-0.166667 3:-0.864407 4:-0.833333
1 1:-0.722222 2:0.166667 3:-0.694915 4:-0.916667
0 1:0.166667 2:-0.416667 3:0.457627 4:0.5
1 1:-0.833333 3:-0.864407 4:-0.916667
2 1:-1.32455e-07 2:-0.166667 3:0.220339 4:0.0833333
2 1:-1.32455e-07 2:-0.333333 3:0.0169491 4:-4.03573e-08

多分类LR模型实现为：

def MultiModel(file_multi: String, spark: SparkSession, model_path: String): Unit = {val training = spark.read.format("libsvm").load(file_multi)val lr = new LogisticRegression().setMaxIter(10).setRegParam(0.3).setElasticNetParam(0.8)// Fit the modelval lrModel = lr.fit(training)// Print the coefficients and intercept for multinomial logistic regressionprintln(s"Coefficients: \n${lrModel.coefficientMatrix}")println(s"Intercepts: ${lrModel.interceptVector}")
}

参考资料

https://spark.apache.org/docs/2.1.0/mllib-linear-methods.html#classification

https://spark.apache.org/docs/2.1.0/ml-classification-regression.html#logistic-regression

https://blog.csdn.net/pupilxmk/article/details/80735599

【技术服务】，详情点击查看： https://mp.weixin.qq.com/s/PtX9ukKRBmazAWARprGIAg

扫一扫关注微信公众号！号主专注于搜索和推荐系统，尝试使用算法去更好的服务于用户，包括但不局限于机器学习，深度学习，强化学习，自然语言理解，知识图谱，还不定时分享技术，资料，思考等文章！

Spark排序算法系列之（MLLib、ML）LR使用方式介绍（模型训练、保存、加载、预测）相关推荐

精通八大排序算法系列：二、堆排序算法
精通八大排序算法系列:二.堆排序算法作者:July .二零一一年二月二十日本文参考:Introduction To Algorithms,second edition. ------------- ...
排序算法系列：Shell 排序算法
概述希尔排序(Shell Sort)是 D.L.Shell 于 1959 年提出来的一种排序算法,在这之前排序算法的时间复杂度基本都是 O(n2n^{2}n2) 的,希尔排序算法是突破这个时间复杂度 ...
八十八、Python | 十大排序算法系列（下篇）
@Author:Runsen @Date:2020/7/10 人生最重要的不是所站的位置,而是内心所朝的方向.只要我在每篇博文中写得自己体会,修炼身心:在每天的不断重复学习中,耐住寂寞,练就真功,不畏 ...
八十七、Python | 十大排序算法系列（上篇）
@Author:Runsen @Date:2020/7/10 人生最重要的不是所站的位置,而是内心所朝的方向.只要我在每篇博文中写得自己体会,修炼身心:在每天的不断重复学习中,耐住寂寞,练就真功,不畏 ...
排序算法系列之（二）——冒泡排序名字最为形象的一个
前言大约在上个冬季我给自己挖了一个坑(想要总结排序~~(>_<)~~),感觉把自己埋起来会暖和一点,可是大约一年过去了,埋的越来越深,却丝毫感觉不到暖意--被我的诗意打动了有没有,已经深 ...
排序算法系列之（七）——分分合合的归并排序
文章目录前言归并排序排序过程代码实现代码分析运行测试总结前言再一次总结基础的排序算法,印象里距离上一次总结排序也没过多久,查询后才发现上一篇总结<排序算法系列之(六)--逐步砍 ...
【排序算法系列 1】冒泡排序（Bubble Sort）
[排序算法系列 1]冒泡排序 [排序算法系列 2]选择排序 [排序算法系列 3] 插入排序 [排序算法系列 4] 高级排序--希尔排序(插入排序的改进) [排序算法系列 5] 高级排序--归并排序 [ ...
【十大排序算法系列】快速排序
写在前面上一篇更新了这个系列的排序算法([十大排序算法系列]冒泡排序).分析了冒泡的逻辑和优化点,下面来写下快速排序(为什么跳这么快?因为比较走心hhhh) 照例给出系列内所有算法的对比.. 常见的 ...
人工智能算法之梯度下降法、协同过滤、相似度技术、ALS算法（附案例分析）、模型存储与加载、推荐系统的冷启动问题
梯度下降法求解机器学习算法的模型参数,即无约束优化问题时,梯度下降法是最常采用的方法之一,另一种常用的方法是最小二乘法.这里对梯度下降法做简要介绍. 最小二乘法法适用于模型方程存在解析解的情况.如果 ...

Spark排序算法系列之（MLLib、ML）LR使用方式介绍（模型训练、保存、加载、预测）

背景