PySpark：LogisticRegressionWithLBFGS 在迭代中变得越来越慢答案

【问题标题】：PySpark: LogisticRegressionWithLBFGS is getting slower in iterationPySpark：LogisticRegressionWithLBFGS 在迭代中变得越来越慢
【发布时间】：2017-07-21 14:09:13
【问题描述】：

我有一个迭代调用 LogisticRegressionWithLBFGS x 次。

问题是，每次循环的迭代都变得越来越慢，最终永远挂起。

我尝试了很多不同的方法，但到目前为止都没有运气。

代码如下：

def getBootsrapedAttribution( iNumberOfSamples, df):

    def parsePoint(line):
        return LabeledPoint(line[2], line[3:])

    aResults = {}
    while x <= iNumberOfSamples:
        print ("## Sample: " + str(x))
        a = datetime.datetime.now()
        dfSample = sampleData(df)
        dfSample.repartition(700)
        parsedData = dfSample.rdd.map(parsePoint)
        parsedData = parsedData.repartition(700)
        parsedData.persist()
        model = LogisticRegressionWithLBFGS.train(parsedData)
        parsedData.unpersist()
        b = datetime.datetime.now()
        print(b-a)
        x+=1

def sampleData(df):
    df = df.repartition(500)
    dfFalse = df.filter('col == 0').sample(False, 0.00035)
    dfTrue = df.filter('col == 1')
    dfSample = dfTrue.unionAll(dfFalse)
    return dfSample


getBootsrapedAttribution(50, df)

输出如下所示：

## Sample: 1
0:00:44.393886

## Sample: 2
0:00:28.403687

## Sample: 3
0:00:30.884087

## Sample: 4
0:00:33.523481

## Sample: 5
0:00:36.107836

## Sample: 6
0:00:37.077169

## Sample: 7
0:00:41.160941

## Sample: 8
0:00:54.768870

## Sample: 9
0:01:01.31139

## Sample: 10
0:00:59.326750

## Sample: 11
0:01:37.222967

## Sample: 12

...hangs forever

没有model = LogisticRegressionWithLBFGS.train(parsedData)，它运行时不会出现性能问题。

我的集群如下所示：

spark.default.parallelism   500
spark.driver.maxResultSize  20G
spark.driver.memory 200G
spark.executor.cores    32
spark.executor.instances    2
spark.executor.memory   124G

有人知道这个问题吗？

【问题讨论】：

标签： apache-spark pyspark logistic-regression

【解决方案1】：

我正在回答我自己的问题。

问题在于方法LogisticRegressionWithLBFGS。在Spark 2.1+ 中用LogisticRegression 替换此方法解决了该问题。每次迭代都不再减速。

此外，您还可以使用上面的代码进行一些改进。 rdd 方法sample 可以替换为DataFrame 方法sampleBy。这也将避免不必要的union：

.sampleBy('col', fractions={0: 0.00035, 1: 1}, seed=1234)

此外，以上代码中的所有repartitions 都是不必要的。重要的是传递给getBootsrapedAttribution() 的df 被很好地分割并且cached。

【讨论】：