SparkSql if value null take previous one

user_2258585 0
Viewed 368

So I have this dataframe

+---+-------------+-----+
| id|    timestamp|  num|
+---+-------------+-----+
| 10|1546300799000| 37.5|
| 10|1546300800000| null|
| 10|1546300801000| null|
| 10|1546300802000|37.51|
| 20|1546300804000| null|
| 10|1546300806000| 37.5|
| 10|1546300807000| null|
+---+-------------+-----+

What I'm trying to achieve is that num must be updated with the value itself if present or if null the "last" value that gets from previous row (ordered by timestamp and grouped by id)

So this should be output

+---+-------------+-----+
| id|    timestamp|  num|
+---+-------------+-----+
| 10|1546300799000| 37.5|
| 10|1546300800000| 37.5|
| 10|1546300801000| 37.5|
| 10|1546300802000|37.51|
| 20|1546300804000| null|
| 10|1546300806000| 37.5|
| 10|1546300807000| 37.5|
+---+-------------+-----+

I came up with this solution

w = Window.partitionBy('id').orderBy('timestamp')
final = joined.withColumn('num2', when(col('num').isNull(), lag(col('num')).over(w)).otherwise(col('num')))

but this is the output I get

+---+-------------+-----+-----+
| id|    timestamp|  num| num2|
+---+-------------+-----+-----+
| 10|1546300799000| 37.5| 37.5|
| 10|1546300800000| null| 37.5|
| 10|1546300801000| null| null|
| 10|1546300802000|37.51|37.51|
| 20|1546300804000| null| null|
| 10|1546300806000| 37.5| 37.5|
| 10|1546300807000| null| 37.5|
+---+-------------+-----+-----+

as you can see a value gets it previous one if isNull, but if you look at the third row, I get a null, and I assume because it gets the value of 2nd row, but when it is still not updated (so still a null from original dataframe).

I'm a little bit lost how should I proceed. Any help?

pyspark apache-spark-sql apache-spark
1 Answers

You are looking to forward fill a measure which unfortunately is not something built in with Pyspark as it is with Pandas. But there is a workaround.

from pyspark.sql import functions as F
from pyspark.sql.window import Window

 window = Window.partitionBy('id')\
           .orderBy('timestamp')\
           .rowsBetween(Window.unboundedPreceding, Window.currentRow)

 final = joined.\
               withColumn('numFilled', F.last('num',ignorenulls = True).over(window)

So what this is doing is that it constructs your window based on the partition key and the order column. It also tells the window to look back for previous rows and up to the current row. Finally, at each row, you return the last value that is not null (which remember according to your window, this includes your current row)

user_27932280
Related