Spark add new column to dataframe with value from previous row

by

You can use lag window function as follows 

from pyspark.sql.functions import lag, col
from pyspark.sql.window import Window

df = sc.parallelize([(4, 9.0), (3, 7.0), (2, 3.0), (1, 5.0)]).toDF(["id", "num"])
w = Window().partitionBy().orderBy(col("id"))
df.select("*", lag("num").over(w).alias("new_col")).na.drop().show()

## +---+---+-------+
## | id|num|new_col|
## +---+---+-------|
## |  2|3.0|    5.0|
## |  3|7.0|    3.0|
## |  4|9.0|    7.0|
## +---+---+-------+

but there some important issues:

  1. if you need a global operation (not partitioned by some other column / columns) it is extremely inefficient.
  2. you need a natural way to order your data.

While the second issue is almost never a problem the first one can be a deal-breaker. If this is the case you should simply convert your DataFrame to RDD and compute lag manually. See for example:

Other useful links:

More Related Contents:

  1. How to add a constant column in a Spark DataFrame?
  2. Spark Dataframe distinguish columns with duplicated name
  3. Pyspark: Split multiple array columns into rows
  4. How do I add a new column to a Spark DataFrame (using PySpark)?
  5. How to change a dataframe column from String type to Double type in PySpark?
  6. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  7. Retrieve top n in each group of a DataFrame in pyspark
  8. Updating a dataframe column in spark
  9. Dividing complex rows of dataframe to simple rows in Pyspark
  10. Filter Pyspark dataframe column with None value
  11. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  12. How to explode multiple columns of a dataframe in pyspark
  13. PySpark: multiple conditions in when clause
  14. Pyspark: Replacing value in a column by searching a dictionary
  15. Count number of non-NaN entries in each column of Spark dataframe in PySpark
  16. Create Spark DataFrame. Can not infer schema for type
  17. Pivot String column on Pyspark Dataframe
  18. Filtering DataFrame using the length of a column
  19. Multiple condition filter on dataframe
  20. How to return a “Tuple type” in a UDF in PySpark?
  21. How to change dataframe column names in pyspark?
  22. How to export a table dataframe in PySpark to csv?
  23. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  24. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  25. Why is Apache-Spark – Python so slow locally as compared to pandas?
  26. Create single row dataframe from list of list PySpark
  27. What is the best way to remove accents with Apache Spark dataframes in PySpark?
  28. How to connect HBase and Spark using Python?
  29. How to pivot on multiple columns in Spark SQL?
  30. Apache Spark — Assign the result of UDF to multiple dataframe columns

Leave a Comment