How to replace negative values with previous positive values in Spark?

Narendra Rokade 0
Viewed 25

I want to replace negative values in spark dataframe with previous positive values. I am using Spark with Java. In Python pandas we are having ffill() api which will help here to solve this issue but in Java it is getting difficult to resolve. I tried using lead/lag function but till where I can check negative values that I am not sure hence this solution will not work.

apache-spark
1 Answers

You can use a Window Function. Take this df as an exemple:

df = spark.createDataFrame(
    [
    ('2018-03-01','6'),
    ('2018-03-02','1'),
    ('2018-03-03','-2'),
    ('2018-03-04','7'),
    ('2018-03-05','-3'),
    ],
    ["date", "value"]
)\
    .withColumn('date', F.col('date').cast('date'))\
    .withColumn('value', F.col('value').cast('integer'))

df.show()

# +----------+-----+
# |      date|value|
# +----------+-----+
# |2018-03-01|    6|
# |2018-03-02|    1|
# |2018-03-03|   -2|
# |2018-03-04|    7|
# |2018-03-05|   -3|
# +----------+-----+

Then, you can use create a column with a when and use a Window Function:

from pyspark.sql import Window

window = Window.partitionBy().rowsBetween(Window.unboundedPreceding, Window.currentRow)

df\
    .withColumn('if<0_than_null', F.when(F.col('value')<0, F.lit(None)).otherwise(F.col('value')))\
    .withColumn('desired_output', F.last('if<0_than_null', ignorenulls=True).over(window))\
    .show()

# +----------+-----+--------------+--------------+
# |      date|value|if<0_than_null|desired_output|
# +----------+-----+--------------+--------------+
# |2018-03-01|    6|             6|             6|
# |2018-03-02|    1|             1|             1|
# |2018-03-03|   -2|          null|             1|
# |2018-03-04|    7|             7|             7|
# |2018-03-05|   -3|          null|             7|
# +----------+-----+--------------+--------------+
Luiz Viola0
Related