[Question] Transform time-series data of a state into time intervals

[doctorseus <do...@devlabs.pro>]

Hi All,

I have a BigQuery dataset which includes event logs from many thousand devices.
It's basically time-series data which includes a "CONNECTED" or
"DISCONNECTED" state for any device at infrequent points in time.
However, except for error cases where it might be missing, there is
always an event logged for any actual state changes and then there are
some events in between just republishing the current state.

Example:

+----------+------------------+---------------------+--------------+
| deviceId | msg seq (device) | timestamp           | state        |
+----------+------------------+---------------------+--------------+
| d111     | 5                | 2023-01-20T20:30:00 | CONNECTED    |
+----------+------------------+---------------------+--------------+
| d111     | 6                | 2023-02-21T01:25:00 | CONNECTED    |
+----------+------------------+---------------------+--------------+
| d000     | 2                | 2023-01-20T10:00:00 | DISCONNECTED |
+----------+------------------+---------------------+--------------+
| d111     | 7                | 2023-02-21T10:15:00 | DISCONNECTED |
<- state change event
+----------+------------------+---------------------+--------------+
| d000     | 3                | 2023-02-21T12:00:00 | DISCONNECTED |
<- state change event
+----------+------------------+---------------------+--------------+
| d000     | 4                | 2023-02-21T12:10:00 | CONNECTED    |
<- state change event
+----------+------------------+---------------------+--------------+
| d000     | 5                | 2023-02-22T05:00:00 | DISCONNECTED |
+----------+------------------+---------------------+--------------+

I would like to use Apache Beam (via Dataflow) to read this dataset
and convert it into a dataset which only lists intervals for the
connected state, similar to this:

+----------+---------------------+---------------------+-----------+
| deviceId | timestamp (start)   | timestamp (end)     | state     |
+----------+---------------------+---------------------+-----------+
| d111     | 2023-01-20T20:30:00 | 2023-02-21T10:15:00 | CONNECTED |
+----------+---------------------+---------------------+-----------+
| d000     | 2023-02-21T12:10:00 | 2023-02-22T05:00:00 | CONNECTED |
+----------+---------------------+---------------------+-----------+

In a naive way I would just loop through all these events, remember
the current state for each device and output an entry each time the
state changes.
But what would be the best way to implement this use-case with Apache
Beam? I was thinking that in theory I should be able to start tracking
state changes for a device at multiple points in time to optimize the
processing.

Given that I need to remember the current state I assume I need some
stateful processing using StateSpec.
But when I read the documentation / blog posts it is also mentioned
that the stateful processing ignores time and doesn't process elements
in order.
So I am quite unsure if there is a good way to actually do this
although I assumed at first that this is a rather trivial task to do
with the available tools.

I assumed this is some common use-case when doing some state analysis
on existing event data, in this case querying the total connection
time.
Is there a best-practice approach for this? Potentially already at the
logging stage of the data to make it easier to process it later.

Thanks!