星火斯卡拉 - 如何計數（）通過兩行空調

Question

我是新手火花斯卡拉，我爲問愚蠢的問題（如果是）而道歉。我被困在一個問題，我簡化如下：

有一個數據框有三列，「machineID」是機器的身份。 「startTime」是任務的開始時間戳。 「endTime」是任務的結束時間戳。

我的目標是計算每臺機器有多少空閒間隔。
例如，
在下表中，第1行和第2行表示在時間0開始並且在時間3結束並且在時間4再次開始的機器＃1，因此時間間隔[3,4]空閒。對於第3行和第4行，機器＃1在時間10開始並在時間20結束，並立即再次開始，因此沒有空閒時間。

machineID, startTime, endTime 
1, 0, 3 
1, 4, 8 
1, 10, 20 
1, 20, 31 
... 
1, 412, 578 
... 
2, 231, 311 
2, 781, 790 
... 

數據幀已經是groupBy（「machineID」）。
我正在使用spark 2.0.1和scala 2.11.8

Answer 1

要訪問DataFrame中的上一行/下一行，我們可以使用Window函數。 在這種情況下，我們將使用lag來訪問上一個結束時間，按machineId分組。

import org.apache.spark.sql.expressions.Window 

// Dataframe Schema 
case class MachineData(id:String, start:Int, end:Int) 
// Sample Data 
machineDF.show 
+---+-----+---+ 
| id|start|end| 
+---+-----+---+ 
| 1| 0| 3| 
| 1| 4| 8| 
| 1| 10| 20| 
| 1| 20| 31| 
| 1| 412|578| 
| 2| 231|311| 
| 2| 781|790| 
+---+-----+---+ 


// define the window as a partition over machineId, ordered by start (time) 
val byMachine = Window.partitionBy($"id").orderBy($"start") 
// we define a new column, "previous end" using the Lag Window function over the previously defined window 
val prevEnd = lag($"end", 1).over(byMachine) 

// new DF with the prevEnd column 
val withPrevEnd = machineDF.withColumn("prevEnd", prevEnd) 
withPrevEnd.show 

+---+-----+---+-------+ 
| id|start|end|prevEnd| 
+---+-----+---+-------+ 
| 1| 0| 3| null| 
| 1| 4| 8|  3| 
| 1| 10| 20|  8| 
| 1| 20| 31|  20| 
| 1| 412|578|  31| 
| 2| 231|311| null| 
| 2| 781|790| 311| 
+---+-----+---+-------+ 

// we're calculating the idle intervals as the numerical diff as an example 
val idleIntervals = withPrevEnd.withColumn("diff", $"start"-$"prevEnd") 
idleIntervals.show 

+---+-----+---+-------+----+ 
| id|start|end|prevEnd|diff| 
+---+-----+---+-------+----+ 
| 1| 0| 3| null|null| 
| 1| 4| 8|  3| 1| 
| 1| 10| 20|  8| 2| 
| 1| 20| 31|  20| 0| 
| 1| 412|578|  31| 381| 
| 2| 231|311| null|null| 
| 2| 781|790| 311| 470| 
+---+-----+---+-------+----+ 

// to calculate the total, we are summing over the differences. Adapt this as your business logic requires. 
val totalIdleIntervals = idleIntervals.select($"id",$"diff").groupBy($"id").agg(sum("diff")) 

+---+---------+ 
| id|sum(diff)| 
+---+---------+ 
| 1|  384| 
| 2|  470| 
+---+---------+