運行時計算不知道開始執行

我想計算多個記錄器的運行時間。可以有無限的錄音機在同一時間運行。運行時計算不知道開始執行

當我有一個開始和結束點，我得到預期的結果與下面的代碼片段。

val ds2 = ds 
     .withColumn("started", when($"status" === "start", 1).otherwise(lit(0))) 
     .withColumn("stopped", when($"status" === "stop", -1).otherwise(lit(0))) 
     .withColumn("engFlag", when($"started" === 1, $"started").otherwise($"stopped")) 
     .withColumn("engWindow", sum($"engFlag").over(Window.orderBy($"timestamp"))) 
     .withColumn("runtime", when($"engWindow" > 0, 
     (unix_timestamp(lead($"timestamp", 1).over(Window.orderBy($"timestamp"))) - unix_timestamp($"timestamp"))/60*$"engWindow").otherwise(lit(0)))

輸入數據：

val ds_working = spark.sparkContext.parallelize(Seq(
     ("2017-01-01 06:00:00", "start", "1"), 
     ("2017-01-01 07:00:00", "start", "2"), 
     ("2017-01-01 08:00:00", "foo", "2"), 
     ("2017-01-01 09:00:00", "blub", "2"), 
     ("2017-01-01 10:00:00", "stop", "3"), 
     ("2017-01-01 11:00:00", null, "3"), 
     ("2017-01-01 12:00:00", "ASC_c", "4"), 
     ("2017-01-01 13:00:00", "stop", "5"), 
     ("2017-01-01 14:00:00", null, "3"), 
     ("2017-01-01 15:00:00", "ASC_c", "4") 
    )).toDF("timestamp", "status", "msg")

輸出：

我的問題

+-------------------+------+---+-------+-------+-------+---------+-------+ 
|   timestamp|status|msg|started|stopped|engFlag|engWindow|runtime| 
+-------------------+------+---+-------+-------+-------+---------+-------+ 
|2017-01-01 06:00:00| start| 1|  1|  0|  1|  1| 60.0| 
|2017-01-01 07:00:00| start| 2|  1|  0|  1|  2| 120.0| 
|2017-01-01 08:00:00| foo| 2|  0|  0|  0|  2| 120.0| 
|2017-01-01 09:00:00| blub| 2|  0|  0|  0|  2| 120.0| 
|2017-01-01 10:00:00| stop| 3|  0|  -1|  -1|  1| 60.0| 
|2017-01-01 11:00:00| null| 3|  0|  0|  0|  1| 60.0| 
|2017-01-01 12:00:00| ASC_c| 4|  0|  0|  0|  1| 60.0| 
|2017-01-01 13:00:00| stop| 5|  0|  -1|  -1|  0| 0.0| 
|2017-01-01 14:00:00| null| 3|  0|  0|  0|  0| 0.0| 
|2017-01-01 15:00:00| ASC_c| 4|  0|  0|  0|  0| 0.0| 
+-------------------+------+---+-------+-------+-------+---------+-------+

現在：
我不知道如何計算運行時間，如果我開始在中間計算一個跑步記錄器。這意味着我沒有看到開始標誌，而是一個停止標誌。這表明過去一開始就必須發生。

數據：

val ds_notworking = spark.sparkContext.parallelize(Seq(
     ("2017-01-01 02:00:00", "foo", "1"), 
     ("2017-01-01 03:00:00", null, "2"), 
     ("2017-01-01 04:00:00", "stop", "1"), 
     ("2017-01-01 05:00:00", "stop", "2"), 
     ("2017-01-01 06:00:00", "start", "1"), 
     ("2017-01-01 07:00:00", "start", "2"), 
     ("2017-01-01 08:00:00", "foo", "2"), 
     ("2017-01-01 09:00:00", "blub", "2"), 
     ("2017-01-01 10:00:00", "stop", "3"), 
     ("2017-01-01 11:00:00", null, "3"), 
     ("2017-01-01 12:00:00", "ASC_c", "4"), 
     ("2017-01-01 13:00:00", "stop", "5"), 
     ("2017-01-01 14:00:00", null, "3"), 
     ("2017-01-01 15:00:00", "ASC_c", "4"), 
    )).toDF("timestamp", "status", "msg")

通緝輸出：

+-------------------+------+---+-------+-------+---------+-----+ 
|   timestamp|status|msg|started|stopped|engWindow|runt | 
+-------------------+------+---+-------+-------+---------+-----+ 
|2017-01-01 02:00:00| foo| 1|  0|  0|  0| 120 | 
|2017-01-01 03:00:00| null| 2|  0|  0|  0| 120 | 
|2017-01-01 04:00:00| stop| 1|  0|  -1|  -1| 60 | 
|2017-01-01 05:00:00| stop| 2|  0|  -1|  -1| 0 | 
|2017-01-01 06:00:00| start| 1|  1|  0|  1| 60 | 
|2017-01-01 07:00:00| start| 2|  1|  0|  1| 120 | 
|2017-01-01 08:00:00| foo| 2|  0|  0|  0| 120 | 
|2017-01-01 09:00:00| blub| 2|  0|  0|  0| 120 | 
|2017-01-01 10:00:00| stop| 3|  0|  -1|  -1| 60 | 
|2017-01-01 11:00:00| null| 3|  0|  0|  0| 60 | 
|2017-01-01 12:00:00| ASC_c| 4|  0|  0|  0| 60 | 
|2017-01-01 13:00:00| stop| 5|  0|  -1|  -1| 0 | 
|2017-01-01 14:00:00| null| 3|  0|  0|  0| 0 | 
|2017-01-01 15:00:00| ASC_c| 4|  0|  0|  0| 0 | 
+-------------------+------+---+-------+-------+---------+-----+

我已經解決了這個問題，當刻錄機只有一個實例可以在同一時間運行：

.withColumn("engWindow", last($"engFlag", true).over(systemWindow.rowsBetween(Window.unboundedPreceding, 0)))

但有2個或更多的實例可悲，我不知道如何做到這一點。如果有人能指引我走向正確的方向，那將會很好。

來源

2017-09-06 user2811630

我想我找到了答案。我正在考慮這種複雜的方式。
雖然我不確定，但如果有任何情況下，這種方法將無法正常工作。

我總結的標誌就像我在工作示例中所做的那樣，通過時間戳降序排列數據，找到最小值並將此值添加到當前值。這應該始終指示正在運行的記錄器的數量。

val ds2 = ds_notworking 
     .withColumn("started", when($"status" === "start", 1).otherwise(lit(0))) 
     .withColumn("stopped", when($"status" === "stop", -1).otherwise(lit(0))) 
     .withColumn("engFlag", when($"started" === 1, $"started").otherwise($"stopped")) 
     .withColumn("engWindow", sum($"engFlag").over(Window.orderBy($"timestamp"))) 
     .withColumn("newEngWindow", $"engWindow" - min($"engWindow").over(Window.orderBy($"timestamp".desc))) 
     .withColumn("runtime2", when($"newEngWindow" > 0, 
     (unix_timestamp(lead($"timestamp", 1).over(Window.orderBy($"timestamp"))) - unix_timestamp($"timestamp"))/60*$"newEngWindow").otherwise(lit(0)))

編輯：也許這將更正確計算minimun值並將其應用到整個窗口。

.withColumn("test1", last(min($"engWindow").over(Window.orderBy($"timestamp"))).over(Window.orderBy($"timestamp").rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)))

輸出：

+-------------------+------+---+-------+-------+-------+---------+------------+--------+ 
|   timestamp|status|msg|started|stopped|engFlag|engWindow|newEngWindow|runtime2| 
+-------------------+------+---+-------+-------+-------+---------+------------+--------+ 
|2017-01-01 02:00:00| foo| 1|  0|  0|  0|  0|   2| 120.0| 
|2017-01-01 03:00:00| null| 2|  0|  0|  0|  0|   2| 120.0| 
|2017-01-01 04:00:00| stop| 1|  0|  -1|  -1|  -1|   1| 60.0| 
|2017-01-01 05:00:00| stop| 2|  0|  -1|  -1|  -2|   0|  0.0| 
|2017-01-01 06:00:00| start| 1|  1|  0|  1|  -1|   1| 60.0| 
|2017-01-01 07:00:00| start| 2|  1|  0|  1|  0|   2| 120.0| 
|2017-01-01 08:00:00| foo| 2|  0|  0|  0|  0|   2| 120.0| 
|2017-01-01 09:00:00| blub| 2|  0|  0|  0|  0|   2| 120.0| 
|2017-01-01 10:00:00| stop| 3|  0|  -1|  -1|  -1|   1| 60.0| 
|2017-01-01 11:00:00| null| 3|  0|  0|  0|  -1|   1| 60.0| 
|2017-01-01 12:00:00| ASC_c| 4|  0|  0|  0|  -1|   1| 60.0| 
|2017-01-01 13:00:00| stop| 5|  0|  -1|  -1|  -2|   0|  0.0| 
|2017-01-01 14:00:00| null| 3|  0|  0|  0|  -2|   0|  0.0| 
|2017-01-01 15:00:00| ASC_c| 4|  0|  0|  0|  -2|   0|  0.0| 
+-------------------+------+---+-------+-------+-------+---------+------------+--------+

來源

2017-09-06 12:00:06 user2811630

運行時計算不知道開始執行

回答

相關問題