title: Filling gaps in timeseries Spark
body: <p>I have a problem dealing with time-series data. Due power failures some timestamps are missing in the dataset. I need to fill this gaps by adding rows, and after that, I can interpolate the missing values. </p>
  
  <p>Input data: </p>
  
  <pre><code>periodstart                usage
  ---------------------------------
  2015-09-11 02:15           23000   
  2015-09-11 03:15           23344   
  2015-09-11 03:30           23283  
  2015-09-11 03:45           23786   
  2015-09-11 04:00           25039
  </code></pre>
  
  <p>Wanted output:</p>
  
  <pre><code>periodstart                usage
  ---------------------------------
  2015-09-11 02:15           23000   
  2015-09-11 02:30           0   
  2015-09-11 02:45           0   
  2015-09-11 03:00           0   
  2015-09-11 03:15           23344   
  2015-09-11 03:30           23283   
  2015-09-11 03:45           23786   
  2015-09-11 04:00           25039  
  </code></pre>
  
  <p>Now I have fixed this with a while loop within a dataset foreach function. The problem is that I have to collect the dataset first to the driver before I can do a while loop. So that is not the right way for Spark. </p>
  
  <p>Can someone give me a better solution?    </p>
  
  <p>this is my code: </p>
  
  <pre><code>MissingMeasurementsDS.collect().foreach(row =&gt; {
    // empty list for new generated measurements
    val output = ListBuffer.empty[Measurement]
    // Missing measurements
    val missingMeasurements = row.getAs[Int]("missingmeasurements")
    val lastTimestamp = row.getAs[Timestamp]("previousperiodstart")
    //Generate missing timestamps
    var i = 1
    while (i &lt;= missingMeasurements) {
      //Increment timestamp with 15 minutes (900000 milliseconds)
      val newTimestamp = lastTimestamp.getTime + (900000 * i)
      output += Measurement(new Timestamp(newTimestamp), 0))
      i += 1
    }
    //Join interpolated measurements with correct measurements
    completeMeasurementsDS.join(output.toDS())
  })
  completeMeasurementsDS.show()
  println("OutputDF count = " + completeMeasurementsDS.count())
  </code></pre>