How to concatenate columns in previous row in dataframe?
19:02 27 May 2020

I have a dataframe like this:

case class CC(id: String, p2: Double, p3: Double, time: Int)
val df = List(
  CC("a", 1.1d, 2.2d, 1),
  CC("b", 3.3d, 4.4d, 2),
  CC("c", 5.5d, 6.6d, 3)).toDF

+---+---+---+----+
| id| p2| p3|time|
+---+---+---+----+
|  a|1.1|2.2|   1|
|  b|3.3|4.4|   2|
|  c|5.5|6.6|   3|
+---+---+---+----+

I want to concatenate p2 and p3 of previous row and place in column p5 and concatenate p2 and p3 of current row and place in column p6. To get:

    +---+---+---+----+---------+---------+
    | id| p2| p3|time| p5      | p6      |
    +---+---+---+----+---------+---------+
    |  a|1.1|2.2|   1|         |1.1: 2.2 |
    |  b|3.3|4.4|   2|1.1: 2.2 |3.3: 4.4 |
    |  c|5.5|6.6|   3|3.3: 4.4 |5.5: 6.6 |
    +---+---+---+----+---------+---------+

For current row, i.e. p6 I can easily use

.withColumn("p6", concat(col("p2"), col("p3")))

and for the previous row, I thought about using a window function and lag like below, but it does not work.

val wf = Window.partitionBy("id").orderBy("time")
df.withColumn("p5", concat(lag(col("p2"), 1) + lag("p3", 1)).over(w))

But I get the error that expression concat... not supported within a window function. Some StackOverflow answers talk about using a user defined aggregate function, but I could not find a simple example that I could follow.

Any explanation on this problem is really appreciated. Please suggest alternate methods to solve this problem if you know. Thanks!

scala dataframe apache-spark functional-programming