Skip to content
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

Jump to bottom

[SPARK-10194] [MLlib] [PySpark] SGD algorithms need convergenceTol parameter in Python #8457

Closed
wants to merge 2 commits into from
Closed

[SPARK-10194] [MLlib] [PySpark] SGD algorithms need convergenceTol parameter in Python #8457

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
92d6d8c
SGD algorithms need convergenceTol parameter in Python
yanboliang Aug 26, 2015
c2f3112
fix typo
yanboliang Sep 11, 2015
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
19 changes: 14 additions & 5 deletions mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,8 @@ private[python] class PythonMLLibAPI extends Serializable {
regParam: Double,
regType: String,
intercept: Boolean,
validateData: Boolean): JList[Object] = {
validateData: Boolean,
convergenceTol: Double): JList[Object] = {
val lrAlg = new LinearRegressionWithSGD()
lrAlg.setIntercept(intercept)
.setValidateData(validateData)
Expand All @@ -141,6 +142,7 @@ private[python] class PythonMLLibAPI extends Serializable {
.setRegParam(regParam)
.setStepSize(stepSize)
.setMiniBatchFraction(miniBatchFraction)
.setConvergenceTol(convergenceTol)
lrAlg.optimizer.setUpdater(getUpdaterFromString(regType))
trainRegressionModel(
lrAlg,
Expand All @@ -159,7 +161,8 @@ private[python] class PythonMLLibAPI extends Serializable {
miniBatchFraction: Double,
initialWeights: Vector,
intercept: Boolean,
validateData: Boolean): JList[Object] = {
validateData: Boolean,
convergenceTol: Double): JList[Object] = {
val lassoAlg = new LassoWithSGD()
lassoAlg.setIntercept(intercept)
.setValidateData(validateData)
Expand All @@ -168,6 +171,7 @@ private[python] class PythonMLLibAPI extends Serializable {
.setRegParam(regParam)
.setStepSize(stepSize)
.setMiniBatchFraction(miniBatchFraction)
.setConvergenceTol(convergenceTol)
trainRegressionModel(
lassoAlg,
data,
Expand All @@ -185,7 +189,8 @@ private[python] class PythonMLLibAPI extends Serializable {
miniBatchFraction: Double,
initialWeights: Vector,
intercept: Boolean,
validateData: Boolean): JList[Object] = {
validateData: Boolean,
convergenceTol: Double): JList[Object] = {
val ridgeAlg = new RidgeRegressionWithSGD()
ridgeAlg.setIntercept(intercept)
.setValidateData(validateData)
Expand All @@ -194,6 +199,7 @@ private[python] class PythonMLLibAPI extends Serializable {
.setRegParam(regParam)
.setStepSize(stepSize)
.setMiniBatchFraction(miniBatchFraction)
.setConvergenceTol(convergenceTol)
trainRegressionModel(
ridgeAlg,
data,
Expand All @@ -212,7 +218,7 @@ private[python] class PythonMLLibAPI extends Serializable {
initialWeights: Vector,
regType: String,
intercept: Boolean,
validateData: Boolean): JList[Object] = {
validateData: Boolean, convergenceTol: Double): JList[Object] = {
Copy link
Contributor

Choose a reason for hiding this comment

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

chop down args

val SVMAlg = new SVMWithSGD()
SVMAlg.setIntercept(intercept)
.setValidateData(validateData)
Expand All @@ -221,6 +227,7 @@ private[python] class PythonMLLibAPI extends Serializable {
.setRegParam(regParam)
.setStepSize(stepSize)
.setMiniBatchFraction(miniBatchFraction)
.setConvergenceTol(convergenceTol)
SVMAlg.optimizer.setUpdater(getUpdaterFromString(regType))
trainRegressionModel(
SVMAlg,
Expand All @@ -240,7 +247,8 @@ private[python] class PythonMLLibAPI extends Serializable {
regParam: Double,
regType: String,
intercept: Boolean,
validateData: Boolean): JList[Object] = {
validateData: Boolean,
convergenceTol: Double): JList[Object] = {
val LogRegAlg = new LogisticRegressionWithSGD()
LogRegAlg.setIntercept(intercept)
.setValidateData(validateData)
Expand All @@ -249,6 +257,7 @@ private[python] class PythonMLLibAPI extends Serializable {
.setRegParam(regParam)
.setStepSize(stepSize)
.setMiniBatchFraction(miniBatchFraction)
.setConvergenceTol(convergenceTol)
LogRegAlg.optimizer.setUpdater(getUpdaterFromString(regType))
trainRegressionModel(
LogRegAlg,
Expand Down
17 changes: 12 additions & 5 deletions python/pyspark/mllib/classification.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ class LogisticRegressionWithSGD(object):
@classmethod
def train(cls, data, iterations=100, step=1.0, miniBatchFraction=1.0,
initialWeights=None, regParam=0.01, regType="l2", intercept=False,
validateData=True):
validateData=True, convergenceTol=0.001):
"""
Train a logistic regression model on the given data.

Expand Down Expand Up @@ -274,11 +274,13 @@ def train(cls, data, iterations=100, step=1.0, miniBatchFraction=1.0,
:param validateData: Boolean parameter which indicates if
the algorithm should validate data
before training. (default: True)
:param convergenceTol: A condition which decides iteration termination.
(default: 0.001)
"""
def train(rdd, i):
return callMLlibFunc("trainLogisticRegressionModelWithSGD", rdd, int(iterations),
float(step), float(miniBatchFraction), i, float(regParam), regType,
bool(intercept), bool(validateData))
bool(intercept), bool(validateData), float(convergenceTol))

return _regression_train_wrapper(train, LogisticRegressionModel, data, initialWeights)

Expand Down Expand Up @@ -439,7 +441,7 @@ class SVMWithSGD(object):
@classmethod
def train(cls, data, iterations=100, step=1.0, regParam=0.01,
miniBatchFraction=1.0, initialWeights=None, regType="l2",
intercept=False, validateData=True):
intercept=False, validateData=True, convergenceTol=0.001):
"""
Train a support vector machine on the given data.

Expand Down Expand Up @@ -472,11 +474,13 @@ def train(cls, data, iterations=100, step=1.0, regParam=0.01,
:param validateData: Boolean parameter which indicates if
the algorithm should validate data
before training. (default: True)
:param convergenceTol: A condition which decides iteration termination.
(default: 0.001)
"""
def train(rdd, i):
return callMLlibFunc("trainSVMModelWithSGD", rdd, int(iterations), float(step),
float(regParam), float(miniBatchFraction), i, regType,
bool(intercept), bool(validateData))
bool(intercept), bool(validateData), float(convergenceTol))

return _regression_train_wrapper(train, SVMModel, data, initialWeights)

Expand Down Expand Up @@ -600,12 +604,15 @@ class StreamingLogisticRegressionWithSGD(StreamingLinearAlgorithm):
:param miniBatchFraction: Fraction of data on which SGD is run for each
iteration.
:param regParam: L2 Regularization parameter.
:param convergenceTol: A condition which decides iteration termination.
Copy link
Contributor

Choose a reason for hiding this comment

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

Not part of this PR, but do you mind documenting default values?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, I think I can do these in the follow-up PR because I found that there are much room to improve for StreamingLogisticRegressionWithSGD of PySpark. Let this PR focus on convergenceTol parameter related issues.

"""
def __init__(self, stepSize=0.1, numIterations=50, miniBatchFraction=1.0, regParam=0.01):
def __init__(self, stepSize=0.1, numIterations=50, miniBatchFraction=1.0, regParam=0.01,
convergenceTol=0.001):
self.stepSize = stepSize
self.numIterations = numIterations
self.regParam = regParam
self.miniBatchFraction = miniBatchFraction
self.convergenceTol = convergenceTol
self._model = None
super(StreamingLogisticRegressionWithSGD, self).__init__(
model=self._model)
Expand Down
32 changes: 21 additions & 11 deletions python/pyspark/mllib/regression.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@
'LinearRegressionModel', 'LinearRegressionWithSGD',
'RidgeRegressionModel', 'RidgeRegressionWithSGD',
'LassoModel', 'LassoWithSGD', 'IsotonicRegressionModel',
'IsotonicRegression']
'IsotonicRegression', 'StreamingLinearAlgorithm',
'StreamingLinearRegressionWithSGD']


class LabeledPoint(object):
Expand Down Expand Up @@ -202,7 +203,7 @@ class LinearRegressionWithSGD(object):
@classmethod
def train(cls, data, iterations=100, step=1.0, miniBatchFraction=1.0,
initialWeights=None, regParam=0.0, regType=None, intercept=False,
validateData=True):
validateData=True, convergenceTol=0.001):
"""
Train a linear regression model using Stochastic Gradient
Descent (SGD).
Expand Down Expand Up @@ -244,11 +245,14 @@ def train(cls, data, iterations=100, step=1.0, miniBatchFraction=1.0,
:param validateData: Boolean parameter which indicates if
the algorithm should validate data
before training. (default: True)
:param convergenceTol: A condition which decides iteration termination.
(default: 0.001)
"""
def train(rdd, i):
return callMLlibFunc("trainLinearRegressionModelWithSGD", rdd, int(iterations),
float(step), float(miniBatchFraction), i, float(regParam),
regType, bool(intercept), bool(validateData))
regType, bool(intercept), bool(validateData),
float(convergenceTol))

return _regression_train_wrapper(train, LinearRegressionModel, data, initialWeights)

Expand Down Expand Up @@ -330,7 +334,7 @@ class LassoWithSGD(object):
@classmethod
def train(cls, data, iterations=100, step=1.0, regParam=0.01,
miniBatchFraction=1.0, initialWeights=None, intercept=False,
validateData=True):
validateData=True, convergenceTol=0.001):
"""
Train a regression model with L1-regularization using
Stochastic Gradient Descent.
Expand Down Expand Up @@ -362,11 +366,13 @@ def train(cls, data, iterations=100, step=1.0, regParam=0.01,
:param validateData: Boolean parameter which indicates if
the algorithm should validate data
before training. (default: True)
:param convergenceTol: A condition which decides iteration termination.
(default: 0.001)
"""
def train(rdd, i):
return callMLlibFunc("trainLassoModelWithSGD", rdd, int(iterations), float(step),
float(regParam), float(miniBatchFraction), i, bool(intercept),
bool(validateData))
bool(validateData), float(convergenceTol))

return _regression_train_wrapper(train, LassoModel, data, initialWeights)

Expand Down Expand Up @@ -449,7 +455,7 @@ class RidgeRegressionWithSGD(object):
@classmethod
def train(cls, data, iterations=100, step=1.0, regParam=0.01,
miniBatchFraction=1.0, initialWeights=None, intercept=False,
validateData=True):
validateData=True, convergenceTol=0.001):
"""
Train a regression model with L2-regularization using
Stochastic Gradient Descent.
Expand Down Expand Up @@ -481,11 +487,13 @@ def train(cls, data, iterations=100, step=1.0, regParam=0.01,
:param validateData: Boolean parameter which indicates if
the algorithm should validate data
before training. (default: True)
:param convergenceTol: A condition which decides iteration termination.
(default: 0.001)
"""
def train(rdd, i):
return callMLlibFunc("trainRidgeModelWithSGD", rdd, int(iterations), float(step),
float(regParam), float(miniBatchFraction), i, bool(intercept),
bool(validateData))
bool(validateData), float(convergenceTol))

return _regression_train_wrapper(train, RidgeRegressionModel, data, initialWeights)

Expand Down Expand Up @@ -636,15 +644,17 @@ class StreamingLinearRegressionWithSGD(StreamingLinearAlgorithm):
After training on a batch of data, the weights obtained at the end of
training are used as initial weights for the next batch.

:param: stepSize Step size for each iteration of gradient descent.
:param: numIterations Total number of iterations run.
:param: miniBatchFraction Fraction of data on which SGD is run for each
:param stepSize: Step size for each iteration of gradient descent.
:param numIterations: Total number of iterations run.
:param miniBatchFraction: Fraction of data on which SGD is run for each
Copy link
Contributor

Choose a reason for hiding this comment

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

Not your fault, but if you make any additional changes can you add a "." at the end of this sentence?

Copy link
Contributor

Choose a reason for hiding this comment

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

Also, the default values should be documented.

iteration.
:param convergenceTol: A condition which decides iteration termination.
"""
def __init__(self, stepSize=0.1, numIterations=50, miniBatchFraction=1.0):
def __init__(self, stepSize=0.1, numIterations=50, miniBatchFraction=1.0, convergenceTol=0.001):
self.stepSize = stepSize
self.numIterations = numIterations
self.miniBatchFraction = miniBatchFraction
self.convergenceTol = convergenceTol
self._model = None
super(StreamingLinearRegressionWithSGD, self).__init__(
model=self._model)
Expand Down