Proposal: Watermarking

Start by inspecting the Streamz Dataframe entry points for windowing and aggregation operations; the issue names no files or tests. Clarify how the event-time column and watermark threshold should be configured, how late events are handled, and which aggregation behaviors define done. Use the timestamp table as the basis for acceptance tests.

This is a feature that exists in other streaming data system with windowing and aggregation functions. Its purpose is to support late arriving data in a window or aggregation. It will require that Streamz becomes semi-aware of the data structure because we will need to specify a column that represents the event time. So, for clarity, there are two timestamps here:

1. The time at which the event enters the data pipeline
2. The time at which the event is created in the source. We'll call this the "event time".

Due to various latencies in a distributed system, an event that should be included into an aggregation arrives too late into the pipeline to be counted. As an example, if you have a window of 5 minutes, but an event that has an event time within those 5 minutes arrives 3 minutes after the window closes, it will not be included in any aggregations.

What watermarking will do is keep the window open for a specified amount of time to include all of the data. So, if we have a 5 minute window and a watermarking threshold of 5 minutes, the window will include all events in the first 5 minutes and all events in the second 5 minutes if the event time belongs to the previous 5 minutes. If the event time is outside of the window, it will be dropped. This may be the key to implementing this, because we may just be able to include all data and then just drop data that is outside of the watermark threshold.

Here is an example of windows of 5 seconds and a watermark threshold of 5 seconds.

I've been spending the last few days trying to figure out where this would fit into Streamz because it seems like Streamz doesn't determine what gets included in a batch. So, I'm thinking this could be implemented a few places.

My current thinking is that the Streamz Dataframe would need new parameters for the watermark threshold time and the event time column. And, when operations like windowing or aggregations are performed, it would take into account the watermarking threshold.

As always, feedback is greatly appreciated here. I'd like to include something like this so that it works for everyone.

Also, let me know if this explanation isn't clear.