Difference between two rows in Spark dataframe

by

You can use Window function if the calculation is fixed as calculating difference between previous months, or calculating between previous two months … etc. For that you can use lag and lead function with Window.

But for that you need to change the date column as below so that it can be ordered.

+-------+------+--------------+------+
|Column1|Date  |Date_Converted|Amount|
+-------+------+--------------+------+
|A      |1-jul |2017-07-01    |1000  |
|A      |1-june|2017-06-01    |2000  |
|A      |1-May |2017-05-01    |2000  |
|A      |1-dec |2017-12-01    |3000  |
|A      |1-Nov |2017-11-01    |2000  |
|B      |1-jul |2017-07-01    |100   |
|B      |1-june|2017-06-01    |300   |
|B      |1-May |2017-05-01    |400   |
|B      |1-dec |2017-12-01    |300   |
+-------+------+--------------+------+

You can find the difference between previous month and current month by doing 

import org.apache.spark.sql.expressions._
val windowSpec = Window.partitionBy("Column1").orderBy("Date_Converted")
import org.apache.spark.sql.functions._
df.withColumn("diff_Amt_With_Prev_Month", $"Amount" - when((lag("Amount", 1).over(windowSpec)).isNull, 0).otherwise(lag("Amount", 1).over(windowSpec)))
   .show(false)

You should have 

+-------+------+--------------+------+------------------------+
|Column1|Date  |Date_Converted|Amount|diff_Amt_With_Prev_Month|
+-------+------+--------------+------+------------------------+
|B      |1-May |2017-05-01    |400   |400.0                   |
|B      |1-june|2017-06-01    |300   |-100.0                  |
|B      |1-jul |2017-07-01    |100   |-200.0                  |
|B      |1-dec |2017-12-01    |300   |200.0                   |
|A      |1-May |2017-05-01    |2000  |2000.0                  |
|A      |1-june|2017-06-01    |2000  |0.0                     |
|A      |1-jul |2017-07-01    |1000  |-1000.0                 |
|A      |1-Nov |2017-11-01    |2000  |1000.0                  |
|A      |1-dec |2017-12-01    |3000  |1000.0                  |
+-------+------+--------------+------+------------------------+

You can increase the lagging position for previous two months as 

df.withColumn("diff_Amt_With_Prev_two_Month", $"Amount" - when((lag("Amount", 2).over(windowSpec)).isNull, 0).otherwise(lag("Amount", 2).over(windowSpec)))
  .show(false)

which will give you 

+-------+------+--------------+------+----------------------------+
|Column1|Date  |Date_Converted|Amount|diff_Amt_With_Prev_two_Month|
+-------+------+--------------+------+----------------------------+
|B      |1-May |2017-05-01    |400   |400.0                       |
|B      |1-june|2017-06-01    |300   |300.0                       |
|B      |1-jul |2017-07-01    |100   |-300.0                      |
|B      |1-dec |2017-12-01    |300   |0.0                         |
|A      |1-May |2017-05-01    |2000  |2000.0                      |
|A      |1-june|2017-06-01    |2000  |2000.0                      |
|A      |1-jul |2017-07-01    |1000  |-1000.0                     |
|A      |1-Nov |2017-11-01    |2000  |0.0                         |
|A      |1-dec |2017-12-01    |3000  |2000.0                      |
+-------+------+--------------+------+----------------------------+

I hope the answer is helpful

More Related Contents:

  1. Spark 2.0 Dataset vs DataFrame
  2. Spark – load CSV file as DataFrame?
  3. How to define partitioning of DataFrame?
  4. Encoder error while trying to map dataframe row to updated row
  5. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  6. How to define schema for custom type in Spark SQL?
  7. How can I change column types in Spark SQL’s DataFrame?
  8. Spark unionAll multiple dataframes
  9. How can I pass extra parameters to UDFs in Spark SQL?
  10. How to split a dataframe into dataframes with same column values?
  11. Dropping a nested column from Spark DataFrame
  12. How to read records in JSON format from Kafka using Structured Streaming?
  13. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  14. How to define a custom aggregation function to sum a column of Vectors?
  15. Spark DataFrame: does groupBy after orderBy maintain that order?
  16. Better way to convert a string field into timestamp in Spark
  17. Filling gaps in timeseries Spark
  18. Spark UDAF with ArrayType as bufferSchema performance issues
  19. Apache Spark how to append new column from list/array to Spark dataframe
  20. How to use COGROUP for large datasets
  21. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  22. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  23. Create new Dataframe with empty/null field values
  24. Perform a typed join in Scala with Spark Datasets
  25. Provide schema while reading csv file as a dataframe in Scala Spark
  26. DataFrame-ified zipWithIndex
  27. Spark: Add column to dataframe conditionally
  28. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  29. Calculate Cosine Similarity Spark Dataframe
  30. How to create DataFrame from Scala’s List of Iterables?

Leave a Comment