apache-spark - Apache Spark 是否不如 Scikit Learn 准确？

Question

我最近一直在尝试了解 Apache Spark 作为 Scikit Learn 的替代品，但在我看来，即使在简单的情况下，Scikit 收敛到准确模型的速度也远远快于 Spark。例如，我使用以下脚本为一个非常简单的线性函数 (z=x+y) 生成了 1000 个数据点:

from random import random

def func(in_vals):
    '''result = x (+y+z+w....)'''
    result = 0
    for v in in_vals:
        result += v
    return result

if __name__ == "__main__":
    entry_count = 1000
    dim_count = 2
    in_vals = [0]*dim_count
    with open("data_yequalsx.csv", "w") as out_file:
        for entry in range(entry_count):
            for i in range(dim_count):
                in_vals[i] = random()
            out_val = func(in_vals)
            out_file.write(','.join([str(x) for x in in_vals]))
            out_file.write(",%s\n" % str(out_val))

然后我运行了以下 Scikit 脚本:

import sklearn
from sklearn import linear_model

import numpy as np

data = []
target = []
with open("data_yequalsx.csv") as inFile:
    for row in inFile:
        vals = row.split(",")
        data.append([float(x) for x in vals[:-1]])
        target.append(float(vals[-1]))

test_samples= len(data)/10

train_data = [0]*(len(data) - test_samples)
train_target = [0]*(len(data) - test_samples)
test_data = [0]*(test_samples)
test_target = [0]*(test_samples)
train_index = 0
test_index = 0
for j in range(len(data)):
    if j >= test_samples:
        train_data[train_index] = data[j]
        train_target[train_index] = target[j]
        train_index += 1
    else:
        test_data[test_index] = data[j]
        test_target[test_index] = target[j]
        test_index += 1

model = linear_model.SGDRegressor(n_iter=100, learning_rate="invscaling", eta0=0.0001, power_t=0.5, penalty="l2", alpha=0.0001, loss="squared_loss")
model.fit(train_data, train_target)
print(model.coef_)
print(model.intercept_)

result = model.predict(test_data)
mse = np.mean((result - test_target) ** 2)
print("Mean Squared Error = %s" % str(mse))

然后是这个 Spark 脚本:(使用 Spark-submit ，没有其他参数)

from pyspark.mllib.regression import LinearRegressionWithSGD, LabeledPoint
from pyspark import SparkContext

sc = SparkContext (appName="mllib_simple_accuracy")

raw_data = sc.textFile ("data_yequalsx.csv", minPartitions=10) #MinPartitions doesnt guarantee that you get that many partitions, just that you wont have fewer than that many partitions
data = raw_data.map(lambda line: [float(x) for x in line.split (",")]).map(lambda entry: LabeledPoint (entry[-1], entry[:-1])).zipWithIndex()
test_samples= data.count()/10

training_data = data.filter(lambda (entry, index): index >= test_samples).map(lambda (lp,index): lp)
test_data = data.filter(lambda (entry, index): index < test_samples).map(lambda (lp,index): lp)

model = LinearRegressionWithSGD.train(training_data, step=0.01, iterations=100, regType="l2", regParam=0.0001, intercept=True)
print(model._coeff)
print(model._intercept)

mse = (test_data.map(lambda lp: (lp.label - model.predict(lp.features))**2 ).reduce(lambda x,y: x+y))/test_samples;
print("Mean Squared Error: %s" % str(mse))

sc.stop ()

奇怪的是，spark 给出的误差比 Scikit 给出的误差大一个数量级(分别为 0.185 和 0.045)，尽管这两个模型具有几乎相同的设置(据我所知)我知道这是使用 SGD 进行很少的迭代，因此结果可能会有所不同，但我不会想到会有如此大的差异或如此大的误差，特别是考虑到异常简单的数据。

<小时/>

我对 Spark 有什么误解吗？是不是配置不正确？当然我应该得到比这更小的错误？

Answer 1

SGD，代表随机梯度下降，是一种在线凸优化算法，因此很难并行化，因为它每次迭代都会进行一次更新(有更智能的变体，例如带有小批量的 SGD，但仍然不是很聪明)适合并行环境。

另一方面，批处理算法，例如 L-BFGS，我建议您与 Spark (LogigisticRegressionWithLBFGS) 一起使用，可以轻松并行化，因为它在每个时期进行迭代(它需要查看所有数据点，计算每个点的损失函数的值和梯度，然后进行聚合计算完整的梯度)。

Python 在单机上运行，​​因此 SGD 性能良好。

顺便说一句，如果你查看 MLlib 代码，scikit learn 的 lambda 相当于 lambda/数据集的大小(mllib 优化 1/n*sum(l_i(x_i,f(y_i)) + lambda 而 scikit learn 则优化 sum(l_i(x_i,f(y_i)) + lambda