复制当前行或滞后行

Question

我有表1

number  type   time
--------------------
1       on     1
1       on     5
1       off    10
1       off    15

我需要将其转换为表2

number  type   time    
--------------------
1       on     1
1       off    2
1       on     5
1       off    10
1       on     14
1       off    15

条件是如果＜code＞$“类型”＝＝＝“打开”

我找到了应该复制的行

val window = Window.partitionBy($"number").orderBy($"time")
df
.withColumn("lag_type", lag($"type", 1, null).over(window))
.withColumn("lag1", $"type" === "ON" && $"lag_type" =!= "OFF")
.withColumn("lag2", $"type" =!= "ON" && $"lag_type" === "OFF")

但不知道如何添加行。特别是如果它们基于滞后值。对于当前的一个，我可能会列出列表

type       time
(on, off)  (14, 15)

并像在这里展示的那样爆炸它们，但是当涉及到滞后行时，我又迷失了。

有什么建议吗？我正在使用火花2.2。案例类可以吗？

Answer 1

如果我正确理解了您的问题，那么您只想为以下情况添加行：lag1或lag2为true。

一种方法是:

//Filter only the rows that needs to be changed : 

val df2 = df.withColumn("lag_type", lag($"type", 1, null).over(window))
.withColumn("lag1", $"type" === "ON" && $"lag_type" =!= "OFF")
.withColumn("lag2", $"type" =!= "ON" && $"lag_type" === "OFF")
.filter( $"lag1"|| $"lag2")

 
//Change the rows based on the values of lag1 and lag2 
//then drop extra columns

val newChangedDf = df2
.withColumn("time",when($"lag1", $"time"+1).otherwise($"time"-1))
.withColumn("type",when($"lag1", lit("OFF")).otherwise(lit("ON")))
.drop("lag_type","lag1","lag2")

//Finally add them to the original df. 

val finalDf = df.union(newChangedDf)

注意:这不处理lag1和lag2都< code>true的情况。请根据您的要求操作上述代码。