我在Spark的MLlib中使用LinearRegressionWithSGD时遇到问题。我用他们的例子从这里开始https://spark.apache.org/docs/latest/mllib-linear-methods.html(使用Python接口)。
在他们的例子中,所有特征几乎都是按比例缩放的,平均值约为0,标准偏差约为1。现在,如果我将其中一个按因子10取消缩放,回归就会中断(给出nans或非常大的系数):
from pyspark.mllib.regression import LabeledPoint, LinearRegressionWithSGD
from numpy import array
# Load and parse the data
def parsePoint(line):
values = [float(x) for x in line.replace(',', ' ').split(' ')]
# UN-SCALE one of the features by a factor of 10
values[3] *= 10
return LabeledPoint(values[0], values[1:])
data = sc.textFile(spark_home+"data/mllib/ridge-data/lpsa.data")
parsedData = data.map(parsePoint)
# Build the model
model = LinearRegressionWithSGD.train(parsedData)
# Evaluate the model on training data
valuesAndPreds = parsedData.map(lambda p: (p.label, model.predict(p.features)))
MSE = valuesAndPreds.map(lambda (v, p): (v - p)**2).reduce(lambda x, y: x + y) / valuesAndPreds.count()
print("Mean Squared Error = " + str(MSE))
print "Model coefficients:", str(model)
所以,我想我需要做特征缩放。如果我进行了预缩放,它是有效的(因为我回到了缩放的特征)。然而,现在我不知道如何在原始空间中获得系数。
from pyspark.mllib.regression import LabeledPoint, LinearRegressionWithSGD
from numpy import array
from pyspark.mllib.feature import StandardScaler
from pyspark.mllib.feature import StandardScalerModel
# Load and parse the data
def parseToDenseVector(line):
values = [float(x) for x in line.replace(',', ' ').split(' ')]
# UN-SCALE one of the features by a factor of 10
values[3] *= 10
return Vectors.dense(values[0:])
# Load and parse the data
def parseToLabel(values):
return LabeledPoint(values[0], values[1:])
data = sc.textFile(spark_home+"data/mllib/ridge-data/lpsa.data")
parsedData = data.map(parseToDenseVector)
scaler = StandardScaler(True, True)
scaler_model = scaler.fit(parsedData)
parsedData_scaled = scaler_model.transform(parsedData)
parsedData_scaled_transformed = parsedData_scaled.map(parseToLabel)
# Build the model
model = LinearRegressionWithSGD.train(parsedData_scaled_transformed)
# Evaluate the model on training data
valuesAndPreds = parsedData_scaled_transformed.map(lambda p: (p.label, model.predict(p.features)))
MSE = valuesAndPreds.map(lambda (v, p): (v - p)**2).reduce(lambda x, y: x + y) / valuesAndPreds.count()
print("Mean Squared Error = " + str(MSE))
print "Model coefficients:", str(model)
这里,我得到了变换空间中的所有系数。现在我该如何到达原始空间?我还有scaler_model,它是StandardScalerModel对象。但我无法从中获得均值或方差。这个类唯一的公共方法是transform,它可以将点从原始空间转换为transform。但我无法逆转。
我刚刚遇到这个问题。如果x
在训练数据中为高(>3),则模型甚至不能学习f(x) = x
。太可怕了。
我认为另一种选择是改变步长,而不是缩放数据。这在SPARK-1859中进行了讨论。转述如下:
在Lipschitz常数
L
上,步长应小于1。对于二次损失和GD,最佳收敛发生在stepSize = 1/(2L)
。Spark在损失函数上有一个(1/n)
乘法器。假设您有
n = 5
数据点,最大的特征值是1500
。所以L = 1500 * 1500 / 5
。最佳收敛发生在CCD_ 9。
最后一个等式甚至没有意义(我们是如何在分子中得到2的?)但我以前从未听说过Lipschitz常数,所以我没有资格修正它。无论如何,我认为我们可以尝试不同的步长,直到它开始工作。
要重新表述您的问题,您需要找到截距I
以及求解方程的系数C_1
和C_2
:Y = I + C_1 * x_1 + C_2 * x_2
(其中x_1
和x_2
未缩放)。
设i
为mllib返回的截距。同样地,设c_1
和c_2
是mllib返回的系数(或权重)。
设m_1
为x_1
的未标度均值,m_2
为x_2
的未标度均值。
设s_1
为x_1
的未标度标准差,s_2
为x_2
的未标度标准差。
然后是C_1 = (c_1 / s_1)
、C_2 = (c_2 / s_2)
和
I = i - c_1 * m_1 / s_1 - c_2 * m_2 / s_2
这可以很容易地扩展到3个输入变量:
CCD_ 30和CCD_ 31
你可以很容易地自己计算
import numpy as np
from pyspark.mllib.stat import Statistics
summary = Statistics.colStats(features)
mean = summary.mean()
std = np.sqrt(features.variance())
这些是你的Scaler使用的相同的均值和std。您可以使用python magicdict验证这一点
print scaler_model.__dict__.get('_java_model').std()
print scaler_model.__dict__.get('_java_model').mean()