我正在寻找有关 Spark 1.6 ML 库中并行 LBFGS 和 OWLQN 算法实现的文档。
我找到了 1.6 的页面:http://spark.apache.org/docs/1.6.1/ml-advanced.html但没有关于并行化的内容
对于 2.0:http://spark.apache.org/docs/2.0.0/ml-advanced.html但仍然没有关于并行化的信息
最后,我阅读了代码[link1]。方法
def train(dataset: DataFrame): LogisticRegressionModel
似乎使用 Breeze 优化模型,但我没有找到调用 Spark 函数的位置(map、flatMap、reduce,...)。
在代码[link2]中,map用于计算次梯度,然后再减少计算梯度。
谢谢
最佳答案
简而言之,Spark 使用 Breeze LBFGS 和 OWLQN 优化算法,并为它们提供了一种在每次迭代时计算成本函数梯度的方法。
例如,Spark 的 LogisticRegression 类利用了 LogisticCostFun 类,该类扩展了 Breeze 的 DiffFunction 特性。该成本函数类实现了calculate
抽象方法,其签名为:
override def calculate(coefficients: BDV[Double]): (Double, BDV[Double])
计算方法利用 LogisticAggregator 类,这是完成实际工作的地方。聚合类定义了两个重要的方法:
def add(instance: Instance): this.type // the gradient update equation is hard-coded here
def merge(other: LogisticAggregator): this.type // just adds other's gradient to the current gradient
add 方法定义了添加单个数据点后更新梯度的方法,merge 方法定义了组合两个单独聚合器的方法。此类被传送到执行器,用于聚合每个数据分区,然后用于将所有分区聚合器组合成单个聚合器。最终聚合器实例保存当前迭代的累积梯度,并用于更新驱动节点上的系数。此过程通过调用 LogisticCostFun
类中的 treeAggregate
来控制:
val logisticAggregator = {
val seqOp = (c: LogisticAggregator, instance: Instance) => c.add(instance)
val combOp = (c1: LogisticAggregator, c2: LogisticAggregator) => c1.merge(c2)
instances.treeAggregate(
new LogisticAggregator(coeffs, numClasses, fitIntercept, featuresStd, featuresMean)
)(seqOp, combOp)
}
你可以这样想得更简单:Breeze 实现了几种不同的优化方法(例如 LBFGS、OWLQN),只需要你告诉优化方法如何计算梯度。 Spark 告诉 Breeze 算法如何通过 LogisticCostFun 类计算梯度。 LogisticCostFun 只是说向每个分区发送一个 LogisticAggregator 实例,收集梯度更新,然后将它们发送回驱动程序上进行组合。
关于apache-spark - 优化器 LBFGS OWLQN 实现,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/38702412/