Skip to content

[SPARK-16008][ML] Remove unnecessary serialization in logistic regression #13729

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sethah wants to merge 5 commits into from
Closed

[SPARK-16008][ML] Remove unnecessary serialization in logistic regression #13729

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2a1e632
remove unnecessary serialization
sethah Jun 16, 2016
86505b7
comments
sethah Jun 17, 2016
ef8fdea
dimension corrections
sethah Jun 17, 2016
96b0a45
spark style
sethah Jun 17, 2016
5d668a6
removing dim
sethah Jun 17, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 29 additions & 28 deletions mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -937,50 +937,47 @@ class BinaryLogisticRegressionSummary private[classification] (
* Two LogisticAggregator can be merged together to have a summary of loss and gradient of
* the corresponding joint dataset.
*
* @param coefficients The coefficients corresponding to the features.
* @param numClasses the number of possible outcomes for k classes classification problem in
* Multinomial Logistic Regression.
* @param fitIntercept Whether to fit an intercept term.
* @param featuresStd The standard deviation values of the features.
* @param featuresMean The mean values of the features.
*/
private class LogisticAggregator(
coefficients: Vector,
private val numFeatures: Int,
numClasses: Int,
fitIntercept: Boolean,
featuresStd: Array[Double],
featuresMean: Array[Double]) extends Serializable {
fitIntercept: Boolean) extends Serializable {

private var weightSum = 0.0
private var lossSum = 0.0

private val coefficientsArray = coefficients match {
case dv: DenseVector => dv.values
case _ =>
throw new IllegalArgumentException(
s"coefficients only supports dense vector but got type ${coefficients.getClass}.")
}

private val dim = if (fitIntercept) coefficientsArray.length - 1 else coefficientsArray.length

private val gradientSumArray = Array.ofDim[Double](coefficientsArray.length)
private val gradientSumArray =
Array.ofDim[Double](if (fitIntercept) numFeatures + 1 else numFeatures)

/**
* Add a new training instance to this LogisticAggregator, and update the loss and gradient
* of the objective function.
*
* @param instance The instance of data point to be added.
* @param coefficients The coefficients corresponding to the features.
* @param featuresStd The standard deviation values of the features.
* @return This LogisticAggregator object.
*/
def add(instance: Instance): this.type = {
def add(
instance: Instance,
coefficients: Vector,
featuresStd: Array[Double]): this.type = {
instance match { case Instance(label, weight, features) =>
require(dim == features.size, s"Dimensions mismatch when adding new instance." +
s" Expecting $dim but got ${features.size}.")
require(numFeatures == features.size, s"Dimensions mismatch when adding new instance." +
s" Expecting $numFeatures but got ${features.size}.")
require(weight >= 0.0, s"instance weight, $weight has to be >= 0.0")

if (weight == 0.0) return this

val localCoefficientsArray = coefficientsArray
val coefficientsArray = coefficients match {
case dv: DenseVector => dv.values
case _ =>
throw new IllegalArgumentException(
s"coefficients only supports dense vector but got type ${coefficients.getClass}.")
}
val localGradientSumArray = gradientSumArray

numClasses match {
Expand All @@ -990,11 +987,11 @@ private class LogisticAggregator(
var sum = 0.0
features.foreachActive { (index, value) =>
if (featuresStd(index) != 0.0 && value != 0.0) {
sum += localCoefficientsArray(index) * (value / featuresStd(index))
sum += coefficientsArray(index) * (value / featuresStd(index))
Copy link
Member

@srowen srowen Jun 17, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I could be missing something, but why are the values being normalized by the stdev but not centered by the mean? before, the means were passed in, so I wonder if something was being overlooked before.

Copy link
Contributor Author

@sethah sethah Jun 17, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe it is by design and not being overlooked (perhaps the mean was also passed in case future versions decided it was necessary to mean center the data). This comment is from MLlib:

     * Here, if useFeatureScaling is enabled, we will standardize the training features by dividing
     * the variance of each column (without subtracting the mean), and train the model in the
     * scaled space. Then we transform the coefficients from the scaled space to the original scale
     * as GLMNET and LIBSVM do.

}
}
sum + {
if (fitIntercept) localCoefficientsArray(dim) else 0.0
if (fitIntercept) coefficientsArray(numFeatures) else 0.0
}
}

Expand All @@ -1007,7 +1004,7 @@ private class LogisticAggregator(
}

if (fitIntercept) {
localGradientSumArray(dim) += multiplier
localGradientSumArray(numFeatures) += multiplier
}

if (label > 0) {
Expand All @@ -1034,8 +1031,8 @@ private class LogisticAggregator(
* @return This LogisticAggregator object.
*/
def merge(other: LogisticAggregator): this.type = {
require(dim == other.dim, s"Dimensions mismatch when merging with another " +
s"LeastSquaresAggregator. Expecting $dim but got ${other.dim}.")
require(numFeatures == other.numFeatures, s"Dimensions mismatch when merging with another " +
s"LeastSquaresAggregator. Expecting $numFeatures but got ${other.numFeatures}.")

if (other.weightSum != 0.0) {
weightSum += other.weightSum
Expand Down Expand Up @@ -1086,13 +1083,17 @@ private class LogisticCostFun(
override def calculate(coefficients: BDV[Double]): (Double, BDV[Double]) = {
val numFeatures = featuresStd.length
val coeffs = Vectors.fromBreeze(coefficients)
val n = coeffs.size
val localFeaturesStd = featuresStd


val logisticAggregator = {
val seqOp = (c: LogisticAggregator, instance: Instance) => c.add(instance)
val seqOp = (c: LogisticAggregator, instance: Instance) =>
c.add(instance, coeffs, localFeaturesStd)
val combOp = (c1: LogisticAggregator, c2: LogisticAggregator) => c1.merge(c2)

instances.treeAggregate(
new LogisticAggregator(coeffs, numClasses, fitIntercept, featuresStd, featuresMean)
new LogisticAggregator(numFeatures, numClasses, fitIntercept)
)(seqOp, combOp)
}

Expand Down