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!