Spark / Scala: fill nan with last good observation
<p>I am using the spark 2.0.1 and want to fill nan values with the last good known value in the column.</p>

<p>The only reference for spark I could find  <a href="https://stackoverflow.com/questions/33621319/spark-scala-forward-fill-with-last-observation">Spark / Scala: forward fill with last observation</a> or <a href="https://stackoverflow.com/questions/36343482/fill-in-null-with-previously-known-good-value-with-pyspark">Fill in null with previously known good value with pyspark</a> which seem to use RDD. </p>

<p>I would rather like to stay in the data frame / dataset world and possible handle multiple nan values.
Is this possible?</p>

<p>My assumption is that the data (initially loaded from e.g. a CSV file is ordered by time and this order is preserved in the distributed setting e.g. filling by close / last good known value is correct. Maybe filling with the previous value is enough as for most records there are no 2 or more nan records in a row. Does this actually hold?
The point is that a </p>

<pre><code>myDf.sort("foo").show
</code></pre>

<p>Would destroy any order e.g. all <code>null</code> values will come first.</p>

<p>A small example:</p>

<pre><code>import java.sql.{ Date, Timestamp }
case class FooBar(foo:Date, bar:String)
val myDf = Seq(("2016-01-01","first"),("2016-01-02","second"),("2016-wrongFormat","noValidFormat"), ("2016-01-04","lastAssumingSameDate"))
         .toDF("foo","bar")
         .withColumn("foo", 'foo.cast("Date"))
         .as[FooBar]
</code></pre>

<p>Results in </p>

<pre><code>+----------+--------------------+
|       foo|                 bar|
+----------+--------------------+
|2016-01-01|               first|
|2016-01-02|              second|
|      null|       noValidFormat|
|2016-01-04|lastAssumingSameDate|
+----------+--------------------+
</code></pre>

<p>I would like to fix the value with the last good known value. How can I achieve this?</p>

<pre><code>+----------+--------------------+
|       foo|                 bar|
+----------+--------------------+
|2016-01-01|               first|
|2016-01-02|              second|
|2016-01-02|       noValidFormat|
|2016-01-04|lastAssumingSameDate|
+----------+--------------------+
</code></pre>

<h1>edit</h1>

<p>in my case, it would be good enough to fill the value from the row above, as there are only very limited faulty values.</p>

<h1>edit2</h1>

<p>I try to add an index column</p>

<pre><code>val myDf = Seq(("2016-01-01", "first"), ("2016-01-02", "second"), ("2016-wrongFormat", "noValidFormat"), ("2016-01-04", "lastAssumingSameDate"))
    .toDF("foo", "bar")
    .withColumn("foo", 'foo.cast("Date"))
    .as[FooBar]
    .withColumn("rowId", monotonically_increasing_id())
</code></pre>

<p>And then fill with the last value.</p>

<pre><code>myDf.withColumn("fooLag", lag('foo, 1) over Window.orderBy('rowId)).show
</code></pre>

<p>But that reads the following warning: 
<em>No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.</em> How could I introduce meaningful partitions?</p>

<pre><code>+----------+--------------------+-----+----------+
|       foo|                 bar|rowId|    fooLag|
+----------+--------------------+-----+----------+
|2016-01-01|               first|    0|      null|
|2016-01-02|              second|    1|2016-01-01|
|      null|       noValidFormat|    2|2016-01-02|
|2016-01-04|lastAssumingSameDate|    3|      null|
+----------+--------------------+-----+----------+
</code></pre>