Reputation: 23
I have a file with event_time field, each record is generated every 30 minutes and indicates how many seconds the event lasted. Example:
Event_time | event_duration_seconds
09:00 | 800
09:30 | 1800
10:00 | 2700
12:00 | 1000
13:00 | 1000
I need to transform consecutive events into only one with its duration time. Output file should look like this:
Event_time_start | event_time_end | event_duration_seconds
09:00 | 11:00 | 5300
12:00 | 12:30 | 1000
13:00 | 13:30 | 1000
Is there a method in Scala Spark to compare a dataframe record with the next One?
I tried with a foreach
loop but is not a good option since it is a huge volume of data to process
Upvotes: 2
Views: 1937
Reputation: 22449
Not a trivial problem, but here's a solution with steps as follows:
event_ts_end
using the java.time
APIlag
for event time from previous rowwhen/otherwise
to generate column event_ts_start
with a null
value if the event time difference from the previous row is 30 minuteslast(event_ts_start, ignoreNulls=true)
to backfill null
s with the last event_ts_start
valueevent_ts_start
to aggregate event_duration
and event_ts_end
First, let's assemble a sample dataset:
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import spark.implicits._
val df = Seq(
(101, "2019-04-01 09:00", 800),
(101, "2019-04-01 09:30", 1800),
(101, "2019-04-01 10:00", 2700),
(101, "2019-04-01 12:00", 1000),
(101, "2019-04-01 13:00", 1000),
(220, "2019-04-02 10:00", 1500),
(220, "2019-04-02 10:30", 2400)
).toDF("event_id", "event_time", "event_duration")
Note that the sample dataset has been slightly generalized to include more than a single event and make event time include date
info to cover cases of an event crossing a given date.
Step 1
:
import java.sql.Timestamp
def get_next_closest(seconds: Int) = udf{ (ts: Timestamp, duration: Int) =>
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
val iter = Iterator.iterate(ts.toLocalDateTime)(_.plusSeconds(seconds)).
dropWhile(_.isBefore(ts.toLocalDateTime.plusSeconds(duration)))
iter.next.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
}
Steps 2 - 5
:
val winSpec = Window.partitionBy("event_id").orderBy("event_time")
val seconds = 30 * 60
df.
withColumn("event_ts", to_timestamp($"event_time", "yyyy-MM-dd HH:mm")).
withColumn("event_ts_end", get_next_closest(seconds)($"event_ts", $"event_duration")).
withColumn("prev_event_ts", lag($"event_ts", 1).over(winSpec)).
withColumn("event_ts_start", when($"prev_event_ts".isNull ||
unix_timestamp($"event_ts") - unix_timestamp($"prev_event_ts") =!= seconds, $"event_ts"
)).
withColumn("event_ts_start", last($"event_ts_start", ignoreNulls=true).over(winSpec)).
groupBy($"event_id", $"event_ts_start").agg(
sum($"event_duration").as("event_duration"), max($"event_ts_end").as("event_ts_end")
).show
// +--------+-------------------+--------------+-------------------+
// |event_id| event_ts_start|event_duration| event_ts_end|
// +--------+-------------------+--------------+-------------------+
// | 101|2019-04-01 09:00:00| 5300|2019-04-01 11:00:00|
// | 101|2019-04-01 12:00:00| 1000|2019-04-01 12:30:00|
// | 101|2019-04-01 13:00:00| 1000|2019-04-01 13:30:00|
// | 220|2019-04-02 10:00:00| 3900|2019-04-02 11:30:00|
// +--------+-------------------+--------------+-------------------+
Upvotes: 2