長い系統RDDを持つ反復コードにより、Apache Sparkでstackoverflowエラーが発生する

Question

私はApache Sparkの初心者です。私は現在、RDDを反復的に更新し、エグゼキュータからドライバに約10KBのデータを収集する機械学習プログラムに取り組んでいます。残念ながら、600回以上の繰り返しを実行するとStackOverFlowエラーが発生します。 以下は私のコードです。 反復数が400を超えると、collectAsMap関数でstackoverflowエラーが発生しました。 indexedDevFとindexedDataは、元の質問indexedRDD（ライブラリはhttps://github.com/amplab/spark-indexedrddを提供されるようAMPLabによって開発された）

breakable{ 
    while(bLow > bHigh + 2*tolerance){ 
    indexedDevF = indexedDevF.innerJoin(indexedData){(id, a, b) => (b, a)}.mapValues(x => (x._2 + alphaHighDiff * broad_y.value(iHigh) * kernel(x._1, dataiHigh) + alphaLowDiff * broad_y.value(iLow) * kernel(x._1, dataiLow))) 
    if (iteration % 50 == 0) { 
      indexedDevF.checkpoint() 
    } 
    indexedDevF.persist() // essential to get correct answer 

    val devFMap = indexedDevF.collectAsMap() //0.5s every time according to local:4040! here will stackoverflow 

    var min_value = Double.PositiveInfinity 
    var max_value = -min_value 
    var min_i = -1 
    var max_i = -1 

    i = 0 
    while(i < m){ 

     if(((y(i) > 0) && (alpha(i) < cEpsilon)) || ((y(i) < 0) && (alpha(i) > epsilon))){ 
      if(devFMap(i) <= min_value){ 
       min_value = devFMap(i) 
       min_i = i 
      } 
     } 

     if(((y(i) > 0) && (alpha(i) > epsilon)) || ((y(i) < 0) && (alpha(i) < cEpsilon))){ 
      if(devFMap(i) >= max_value){ 
       max_value = devFMap(i) 
       max_i = i 
      } 
     } 
     i = i+1 
    } 

    iHigh = min_i 
    iLow = max_i 
    bHigh = devFMap(iHigh) 
    bLow = devFMap(iLow) 

    dataiHigh = indexedData.get(iHigh.toLong).get 
    dataiLow = indexedData.get(iLow.toLong).get 

    eta = 2 - 2 * kernel(dataiHigh, dataiLow) 

    alphaHighOld = alpha(iHigh) 
    alphaLowOld = alpha(iLow) 
    var alphaDiff = alphaLowOld - alphaHighOld 
    var lowLabel = y(iLow) 
    var sign = y(iHigh) * lowLabel 

    var alphaLowLowerBound = 0D 
    var alphaLowUpperBound = 0D 

    if (sign < 0){ 
     if (alphaDiff < 0){ 
      alphaLowLowerBound = 0; 
      alphaLowUpperBound = cost + alphaDiff; 
     } 
     else{ 
      alphaLowLowerBound = alphaDiff; 
      alphaLowUpperBound = cost; 
     } 
    } 
    else{ 
     var alphaSum = alphaLowOld + alphaHighOld; 
     if (alphaSum < cost){ 
      alphaLowUpperBound = alphaSum; 
      alphaLowLowerBound = 0; 
     } 
     else{ 
      alphaLowLowerBound = alphaSum - cost; 
      alphaLowUpperBound = cost; 
     } 
    } 

    if (eta > 0){ 
     alphaLowNew = alphaLowOld + lowLabel*(bHigh - bLow)/eta; 
     if (alphaLowNew < alphaLowLowerBound) 
      alphaLowNew = alphaLowLowerBound; 
     else if (alphaLowNew > alphaLowUpperBound) 
      alphaLowNew = alphaLowUpperBound; 
    } 
    else{ 
     var slope = lowLabel * (bHigh - bLow); 
     var delta = slope * (alphaLowUpperBound - alphaLowLowerBound); 
     if (delta > 0){ 
      if (slope > 0) 
       alphaLowNew = alphaLowUpperBound; 
      else 
       alphaLowNew = alphaLowLowerBound; 
     } 
     else 
      alphaLowNew = alphaLowOld; 
    } 

    alphaLowDiff = alphaLowNew - alphaLowOld; 
    alphaHighDiff = -sign*(alphaLowDiff); 
    alpha(iLow) = alphaLowNew; 
    alpha(iHigh) = (alphaHighOld + alphaHighDiff); 


    if(iteration % 50 == 0) 
     print(".") 

    iteration = iteration + 1; 


} 

===================

ある 次のようなものですが、私はチェックポイントが役に立たず、プログラムがスタックオーバーフローエラーで終了することがわかります！私は自分の問題を記述するための簡単なコードを書く。幸いにも、素敵な男が私が問題を解決するのを助けます、あなたは以下の答えを見つけることができます！しかし、チェックポイントが実際に動作する、私はまだ私のプログラム:(RDD.checkpointのドキュメントを見てみると

for(i <- 1 to 1000){ 
    a = a.map(x => x+1).persist 
    var b = a.collect() 
    if(i%100 == 0){ 
    a.checkpoint() 
    } 
    print(".") 
} 

Answer 1

でstackoverflowのエラーを取得し、それは言う：

任意のジョブが実行される前にこの関数が呼び出される必要がありますこのRDDの

そして実際、あなたは少し自分のコードを変更した場合、チェックポイントはaを収集前を行っているために - それは動作しますいいえStackOverflowError：

for(i <- 1 to 1000){ 
    a = a.map(x => x+1).persist 

    if(i%100 == 0){ 
    a.checkpoint() 
    } 

    var b = a.collect() 

    print(".") 
}