title: How to calculate efficiently the index of an array, where the cumulative sum exceeded a threshold in Scala - Spark?
body: <p>Suppose we have an array of integers, of 100 elements.</p>
  
  <p><code>val a = Array(312, 102, 95, 255, ...)</code></p>
  
  <p>I want to find the index (say <code>k</code>) of the array where the cumulative sum of the first <code>k+1</code> elements is greater than a certain threshold, but for the first <code>k</code> element is less.</p>
  
  <p>Because of the high number of elements in the array, I estimated a lower and upper index where in the between the <code>k</code> index should be:</p>
  
  <p><code>k_Lower &lt;= k &lt;= k_upper</code></p>
  
  <p>My question is, what is the best way to find this <code>k</code> index?</p>
  
  <p>I tried it with a <code>while loop</code> when <code>k_lower = 30; k_upper = 47</code> and the <code>threshold = 20000</code></p>
  
  <pre><code>var sum = 0
  var k = 30
  while (k &lt;= 47 &amp;&amp; sum &lt;= 20000) {
    sum = test.take(k).sum
    k += 1
  }
  
  print(k-2)
  </code></pre>
  
  <p>I obtained the right answer, but I'm pretty sure that there is a more efficient or "Scala-ish" solution for this and I'm really new in Scala. Also I have to implement this in Spark.</p>
  
  <p>Another question:
  To optimize the method of finding the <code>k</code> index, I thought of using binary search, where the <code>min</code> and <code>max</code> values are <code>k_lower</code>, respective <code>k_upper</code>. But my attempt to implement this was unsuccessful. How should I do this?</p>
  
  <p>I am using Scala 2.10.6 and Spark 1.6.0</p>
  
  <p><strong>Update!</strong></p>
  
  <p>I thought this approach is a good solution for my problem, but now I think, that I approached it wrongly. My actual problem is the following:</p>
  
  <p>I have a bunch of JSON-s, which are loaded into Spark as an RDD with</p>
  
  <p><code>val eachJson = sc.textFile("JSON_Folder/*.json")</code></p>
  
  <p>I want to split the data into several partitions based on their size. The concatenated JSON-s size should be under a threshold. My idea was to go through the RDD one by one, calculate the size of a JSON and add it to an accumulator. When the accumulator is greater than the threshold, then I remove the last JSON and I obtain a new RDD with all the JSON-s until that, and I do it again with the remaining JSON-s. I read about tail recursion which could be a solution for this, but I wasn't able to implement it, so I tried to solve it differently. I map-ed the sizes for each JSON, and I obtained and RDD[Int]. And I managed to get all the indexes of this array, where the cumulative sum exceeded the threshold:</p>
  
  <pre><code>def calcRDDSize(rdd: RDD[String]): Long = {
    rdd.map(_.getBytes("UTF-8").length.toLong)
      .reduce(_ + _) //add the sizes together
  }
  
  val jsonSize = eachJson.map(s =&gt; s.getBytes("UTF-8").length)
  val threshold = 20000
  val totalSize = calcRDDSize(eachJson)
  val numberOfPartitions = totalSize/threshold
  
  val splitIndexes = scala.collection.mutable.ArrayBuffer.empty[Int]
  var i = 0
  while (i &lt; numberOfPartitions)
    {
      splitIndexes += jsonSize.collect().toStream.scanLeft(0){_ + _}.takeWhile(_ &lt; (i+1)*threshold).length-1
      i = i + 1
    }
  </code></pre>
  
  <p>However, I don't like this solution, because in the while loop I go through several times on the Stream and this is not really efficient. And now I have the indexes where I have to split the RDD, but I don't know how to split is.</p>