Splits a flow into windows that are materialized at certain triggers.

Windows allow developers to split data so we can understand incoming data as time progresses. Once a window is created, we can specify triggers that allow us to customize when the data accumulated on every window is materialized.

Windows must be created by calling one of the window type functions. The supported window types are as follows:

• Global windows - that’s the default window which means all data belongs to one single window. In other words, the data is not split in any way. The window finishes when all producers notify there is no more data

• Fixed windows - splits incoming events into periodic, non- overlapping windows based on event times. In other words, a given event belongs to a single window. If data arrives late, a configured lateness can be specified.

• Periodic windows - splits incoming events into periodic, non- overlapping windows based on processing times. Similar to fixed windows, a given event belongs to a single window.

• Count windows - splits incoming events based on a count. Similar to fixed windows, a given event belongs to a single window.

• Session windows - splits incoming events into unique windows which is grouped until there is a configured gap between event times. Sessions are useful for data that is irregularly distributed with respect to time.

We discuss all types and include examples below. In the first section, “Global windows”, we build the basic intuition about windows and triggers as well as discuss the distinction between “Event time and processing time”. Then we explore “Fixed windows” and the concept of lateness before moving on to other window types.

Global windows

By default, all events belong to the global window. The global window is automatically attached to a partition if no window is specified. The flow below:

Flow.from_stages([some_producer])
|> Flow.partition()
|> Flow.reduce(fn -> 0 end, & &1 + 2)

is equivalent to:

Flow.from_stages([some_producer])
|> Flow.partition(window: Flow.Window.global())
|> Flow.reduce(fn -> 0 end, & &1 + 2)

Even though the global window does not split the data in any way, it already provides conveniences for working with both bounded (finite) and unbounded (infinite) via triggers.

For example, the flow below uses a global window with a count-based trigger to emit the values being summed as we sum them:

iex> window = Flow.Window.global() |> Flow.Window.trigger_every(10)
iex> flow = Flow.from_enumerable(1..100) |> Flow.partition(window: window, stages: 1)
iex> flow |> Flow.reduce(fn -> 0 end, &(&1 + &2)) |> Flow.emit(:state) |> Enum.to_list()
[55, 210, 465, 820, 1275, 1830, 2485, 3240, 4095, 5050, 5050]

Let’s explore the types of triggers available next.

Triggers

Triggers allow us to check point the data processed so far. There are different triggers we can use:

• Event count triggers - compute state operations every X events

• Processing time triggers - compute state operations every X time units for every stage

• Punctuation - hand-written triggers based on the data

Flow supports the triggers above via the trigger_every/3, trigger_periodically/4 and trigger/3 respectively.

Once a trigger is emitted, the reduce/3 step halts and invokes the on_trigger/2 callback, allowing you to emit events and change the reducer accumulator.

Event time and processing time

Before we move to other window types, it is important to discuss the distinction between event time and processing time. In particular, triggers created with the trigger_periodically/4 function are intrinsically inaccurate and therefore should not be used to split the data. For example, if you are measuring the frequency that events arrive, using the event time will always yield the same result, while processing time will be vulnerable to fluctuations if, for instance, an external factor causes events to processed slower or faster than usual.

Furthermore, periodic triggers are established per partition and are message-based, which means partitions will emit the triggers at different times and possibly with delays based on the partition message queue size. However, it is exactly this lack of precision which makes them efficient for checkpointing data.

Flow provides other window types, such as fixed windows, exactly to address the issues with processing time. Such windows use the event time which is based on the data itself. When working with event time, we can assign the data into proper windows even when late or out of order. Such windows can be used to gather time-based insight from the data (for example, the most popular hashtags in the last 10 minutes) as well as for checkpointing data.

Fixed windows (event time)

Fixed windows group the data based on the event times. Regardless if the data is bounded or not, fixed windows give us time-based insight about the data.

Fixed windows are created via the fixed/3 function which specified the duration of the window and a function that retrieves the event time from each event:

Flow.Window.fixed(1, :hour, fn {word, timestamp} -> timestamp end)

Let’s see an example that will use the window above to count the frequency of words based on windows that are 1 hour long. The timestamps used by Flow are integers in milliseconds. For now, we will also set the concurrency down 1 and max demand down to 5 as it is simpler to reason about the results:

iex> data = [{"elixir", 0}, {"elixir", 1_000}, {"erlang", 60_000},
...>         {"concurrency", 3_200_000}, {"elixir", 4_000_000},
...>         {"erlang", 5_000_000}, {"erlang", 6_000_000}]
iex> window = Flow.Window.fixed(1, :hour, fn {_word, timestamp} -> timestamp end)
iex> flow = Flow.from_enumerable(data, max_demand: 5, stages: 1)
iex> flow = Flow.partition(flow, window: window, stages: 1)
iex> flow = Flow.reduce(flow, fn -> %{} end, fn {word, _}, acc ->
...>   Map.update(acc, word, 1, & &1 + 1)
...> end)
iex> flow |> Flow.emit(:state) |> Enum.to_list
[%{"elixir" => 2, "erlang" => 1, "concurrency" => 1},
 %{"elixir" => 1, "erlang" => 2}]

Since the data has been broken in two windows, the first four events belong to the same window while the last 3 belongs to the second one. Notice that reduce/3 is executed per window and that each event belongs to a single window exclusively.

Similar to global windows, fixed windows can also have triggers, allowing us to checkpoint the data as the computation happens.

Data ordering, watermarks and lateness

When working with event time, Flow assumes by default that events are time ordered. This means that, when we move from one window to another, like when we received the entry {"elixir", 4_000_000} in the example above, we assume the previous window has been completed.

Let’s change the events above to be out of order and move the first event to the end of the dataset and see what happens:

iex> data = [{"elixir", 1_000}, {"erlang", 60_000},
...>         {"concurrency", 3_200_000}, {"elixir", 4_000_000},
...>         {"erlang", 5_000_000}, {"erlang", 6_000_000}, {"elixir", 0}]
iex> window = Flow.Window.fixed(1, :hour, fn {_word, timestamp} -> timestamp end)
iex> flow = Flow.from_enumerable(data) |> Flow.partition(window: window, stages: 1, max_demand: 5)
iex> flow = Flow.reduce(flow, fn -> %{} end, fn {word, _}, acc ->
...>   Map.update(acc, word, 1, & &1 + 1)
...> end)
iex> flow |> Flow.emit(:state) |> Enum.to_list
[%{"elixir" => 1, "erlang" => 1, "concurrency" => 1},
 %{"elixir" => 1, "erlang" => 2}]

Notice that now the first map did not count the “elixir” word twice. Since the event arrived late, it was marked as lost. However, in many flows we actually expect data to arrive late or out of order, especially when talking about concurrent data processing.

Luckily, event time windows include the concept of lateness, which is a processing time base period we would wait to receive late events. Let’s change the example above once more but now change the window to also call allowed_lateness/4:

iex> data = [{"elixir", 1_000}, {"erlang", 60_000},
...>         {"concurrency", 3_200_000}, {"elixir", 4_000_000},
...>         {"erlang", 5_000_000}, {"erlang", 6_000_000}, {"elixir", 0}]
iex> window = Flow.Window.fixed(1, :hour, fn {_word, timestamp} -> timestamp end)
iex> window = Flow.Window.allowed_lateness(window, 5, :minute)
iex> flow = Flow.from_enumerable(data) |> Flow.partition(window: window, stages: 1, max_demand: 5)
iex> flow = Flow.reduce(flow, fn -> %{} end, fn {word, _}, acc ->
...>   Map.update(acc, word, 1, & &1 + 1)
...> end)
iex> flow |> Flow.emit(:state) |> Enum.to_list
[%{"concurrency" => 1, "elixir" => 1, "erlang" => 1},
 %{"concurrency" => 1, "elixir" => 2, "erlang" => 1},
 %{"elixir" => 1, "erlang" => 2}]

Now that we allow late events, we can see the first window emitted twice. Instead of the window being marked as done when 1 hour passes, we say it emits a watermark trigger. The window will be effectively done only after the allowed lateness period. If desired, we can use Flow.on_trigger/2 to get more information about each particular window and its trigger. Replace the last line above by the following:

flow
|> Flow.on_trigger(fn state, _index, trigger -> {[{state, trigger}], state} end)
|> Enum.to_list()

The trigger parameter will include the type of window, the current window and what caused the window to be emitted (:watermark or :done).

Note that all stages must receive an event that is outside of a specific window before that window is considered complete. In other words if there are multiple stages in the partition preceding a reduce operation that has a window, the reduce step won’t release a window until it has seen an event that is outside of that window from all processes that it receives data from. This could have an effect on how long events are delayed in the reduce step.

Periodic windows (processing time)

Periodic windows are similar to fixed windows except triggers are emitted based on processing time instead of event time. Remember that relying on periodic windows or triggers is intrinsically inaccurate and should not be used to split the data, only as a checkpointing device.

Periodic windows are also similar to global windows that use trigger_periodically/2 to emit events periodically. The difference is that periodic windows emit a window in a given interval while a trigger emits a trigger. This behaviour may affect functions such as Flow.departition/4, which calls the merge callback per trigger but the done callback per window. Unless you are relying on functions such as Flow.departition/4, there is no distinction between periodic windows and global windows with periodic triggers.

Count windows (event count)

Count windows are simpler versions of fixed windows where windows are split apart by event count. Since it is not timed-based, it does not provide the concept of lateness.

iex> window = Flow.Window.count(10)
iex> flow = Flow.from_enumerable(1..100) |> Flow.partition(window: window, stages: 1)
iex> flow |> Flow.reduce(fn -> 0 end, &(&1 + &2)) |> Flow.emit(:state) |> Enum.to_list()
[55, 155, 255, 355, 455, 555, 655, 755, 855, 955, 0]

Count windows are also similar to global windows that use trigger_every/2 to emit events per count. The difference is that count windows emit a window per event count while a trigger belongs to a window. This behaviour may affect functions such as Flow.departition/4, which calls the merge callback per trigger but the done callback per window. Unless you are relying on functions such as Flow.departition/4, there is no distinction between count windows and global windows with count triggers.