spark mllib 分散SGD実装調査メモ

をここに書いていく

- 分散でどうやってgradient updateやっているか
 - workerの結果を平均している。
 - どこで、broadcastで配って, driverで平均とっている
 - shuffleで平均している？
- 学習係数はどんな感じ？
 - AdaGradとかに更新できそうか？
- BSPとの違いは？
- SSPにするには何がいる？
- 非効率って言われているけど、よさげにできるか？
- いろいろimprovementが提案、実装されているけど、それってどうなの
- LRでの SGD, L-BFGSの比較
 - 前の実験結果だとLR-SGDは精度がでてなかった。十分反復させて収束するかみる。分散環境で
- AdaGrad, ADAMをプロトタイプしてみて目的関数の下がり具合を見てみる。するなら、今週末しかない。


なので、書くことは

- 現状のSGD実装のソースコードリーディング
- jiraチケットの関連話題の紹介
 - 特にSGDの効率化、高速化のチケット
- LRでのSGD, L-BFGSとの収束率の測定
 - ソースいじって、iter毎の目的関数値を出力させるようにする。

- AdaGrad, ADAMプロトタイピング + 実験

最後までいけたら良いな。。。


jira tickect links

- https://issues.apache.org/jira/browse/SPARK-6346?filter=12332858

- https://issues.apache.org/jira/browse/SPARK-1270?filter=12332858

- https://issues.apache.org/jira/browse/SPARK-1359?filter=12332858

論文 links 

- http://learningsys.org/papers/LearningSys_2015_paper_17.pdf

```
@DeveloperApi
class L1Updater extends Updater {
  override def compute(
      weightsOld: Vector,
      gradient: Vector,
      stepSize: Double,
      iter: Int,
      regParam: Double): (Vector, Double) = {
    val thisIterStepSize = stepSize / math.sqrt(iter)　※
    // Take gradient step
    val brzWeights: BV[Double] = weightsOld.toBreeze.toDenseVector
    brzAxpy(-thisIterStepSize, gradient.toBreeze, brzWeights)
```

thisIterStepSizeは※

```file:GradientDescent.scala

/**
     * For the first iteration, the regVal will be initialized as sum of weight squares
     * if it's L2 updater; for L1 updater, the same logic is followed.
     */
    var regVal = updater.compute(
      weights, Vectors.zeros(weights.size), 0, 1, regParam)._2

    var converged = false // indicates whether converged based on convergenceTol
    var i = 1
    while (!converged && i <= numIterations) {
      val bcWeights = data.context.broadcast(weights)
      // Sample a subset (fraction miniBatchFraction) of the total data
      // compute and sum up the subgradients on this subset (this is one map-reduce)
      val (gradientSum, lossSum, miniBatchSize) = data.sample(false, miniBatchFraction, 42 + i)
        .treeAggregate((BDV.zeros[Double](n), 0.0, 0L))(
          seqOp = (c, v) => {
            // c: (grad, loss, count), v: (label, features)
            val l = gradient.compute(v._2, v._1, bcWeights.value, Vectors.fromBreeze(c._1))
            (c._1, c._2 + l, c._3 + 1)
          },
          combOp = (c1, c2) => {
            // c: (grad, loss, count)
            (c1._1 += c2._1, c1._2 + c2._2, c1._3 + c2._3)
          })

      if (miniBatchSize > 0) {
        /**
         * NOTE(Xinghao): lossSum is computed using the weights from the previous iteration
         * and regVal is the regularization value computed in the previous iteration as well.
         */
        stochasticLossHistory.append(lossSum / miniBatchSize + regVal)
        val update = updater.compute(
          weights, Vectors.fromBreeze(gradientSum / miniBatchSize.toDouble),
          stepSize, i, regParam)
        weights = update._1
        regVal = update._2

        previousWeights = currentWeights
        currentWeights = Some(weights)
        if (previousWeights != None && currentWeights != None) {
          converged = isConverged(previousWeights.get,
            currentWeights.get, convergenceTol)
        }
      } else {
        logWarning(s"Iteration ($i/$numIterations). The size of sampled batch is zero")
      }
      i += 1
    }

    logInfo("GradientDescent.runMiniBatchSGD finished. Last 10 stochastic losses %s".format(
      stochasticLossHistory.takeRight(10).mkString(", ")))

    (weights, stochasticLossHistory.toArray)

  }

```

step sizeはdefault 1.0,

stepSize = stepSize / sqrt(iteration)

で減衰
```
/**
   * Set the initial step size of SGD for the first step. Default 1.0.
   * In subsequent steps, the step size will decrease with stepSize/sqrt(t)
   */
  def setStepSize(step: Double): this.type = {
    this.stepSize = step
    this
  }
```


updaterは

- gradient, weightvectorを引数にとって、weightvectorを更新して、L1,L2の正則化項を計算する。

gradientは

- exampleとcurrent weightvectorを引数にとってgradientを計算して、Lossを計算して返す。


treeAggregationは

- partition毎にaggregateする。
- さらにscale個のpartionをaggregateして、新しい numPartition/scaleのpartiallyAggregated partionを作る
- これを繰り返し scale ＋ numPartition /scale以下になったら、最後に一回aggregateして、結果を返す。
- 目的は、一発のaggregationはreducerが１つになって処理が1workerに偏るのを防ぐために、tree構造で段階的、分散でにaggregationする。


stochasticLossHistory.append(lossSum / miniBatchSize + regVal)

でlossの履歴を追加。minBatchSizeで割っているのは一応全体のロスにするため？

収束判定は

- lossで見てない
- weighvectorの変化がconvergenceTolelance以下だったら収束している判断


- runMiniBatchSGDははweightvectorのlossの変化の履歴を返す。
- なので、終わったあとに全lossの履歴を表示、プロットすれば収束率はだせる。
- これを使った、AdaGrad, ADAMで収束率を比較できる


- treeaggregateは reduceと同じなので、毎回、更新したweightvectorはdriverに帰ってきて、それを毎回broadecastで配っている。
- driver側で、adagradでweightvectorの成分を変化させて、broadcastで配るのは簡単
- adamも同じだろうけど、必要な情報がちょっと増える。
- なんである程度簡易で実装できるが、GradientDescent.scalaでなくて
 - AdaGrad.scalaとかADAM.scalaとか作って処理させるのが良さそう