Flow Documentation

Flow requires Elixir v1.7 and Erlang/OTP 22+. To install, add :flow to your dependencies in your project's mix.exs file.

def deps do
  [{:flow, "~> 1.0"}]
end

Because Flow pipelines start several processes linked to the current process, an error in a Flow computation will shut down the Livebook runtime. To prevent this, use one of the following two methods:

Method 1: Use non-linked streams

Convert the Flow to a non-linked stream using Flow.stream(flow, link: false). You can then use standard Enum or Stream functions.

Method 2: Trap exits

Set the process to trap exits before starting the Flow computation using Process.flag(:trap_exit, true).

# Method 1: Non-linked stream
Flow.from_enumerable([1, 2, 3])
|> Flow.map(& &1 * 2)
|> Flow.stream(link: false)
|> Enum.to_list()

# Method 2: Trapping exits
Process.flag(:trap_exit, true)

The Flow.Window.Fixed reducer operates in three distinct stages to manage event grouping and window emission:

1. Event Grouping & Reduction: Incoming events are grouped into windows based on the by function and the duration. Events belonging to the same window are reduced using the provided reducer_fun.
2. Producer Tracking: The system tracks the most recent timestamp/window for each producer and identifies the minimum and maximum windows seen across all producers.
3. Window Catch-up & Triggering: The system compares the minimum window seen by all producers against the current state. It 'catches up' the global window to the minimum seen window, emitting triggers for older windows to ensure data completeness across all producers.

Flow allows you to express computations on collections that execute in parallel using multiple GenStages. This example demonstrates how to stream a file, split it into words, partition the work, and reduce it into a frequency map in parallel.

File.stream!("path/to/some/file")
|> Flow.from_enumerable()
|> Flow.flat_map(&String.split(&1, " "))
|> Flow.partition()
|> Flow.reduce(fn -> %{} end, fn word, acc ->
  Map.update(acc, word, 1, & &1 + 1)
end)
|> Enum.to_list()

The Flow.Window.Fixed module implements a windowing strategy based on a fixed duration. When using this strategy, you must provide a configuration struct with the following keys:

• by: A function that extracts an integer (representing a timestamp or sequence) from an event to determine its window.
• duration: An integer representing the fixed size of each window.
• trigger: A function used to handle window triggers (e.g., :done, :watermark, or :placeholder).
• lateness: (Optional) An integer specifying how much lateness to allow. Defaults to 0.
• periodically: (Optional) A list of periodic triggers.

Note: The by function must return an integer. If it returns any other type, the process will raise an error.

The reducer_fun is responsible for aggregating data within a window. It supports two different arities depending on whether you need to handle window triggers:

• 4-arity: reducer_fun(ref, events, window_acc, index) Used when you do not need to handle explicit window triggers.
• 5-arity: reducer_fun(ref, events, window_acc, index, trigger) Used when you want to handle window triggers. The trigger argument will receive a tuple in the format {:fixed, window_start_time, trigger_type}, where trigger_type can be :placeholder, :watermark, or :done.

The Flow.Window.Global module implements a windowing strategy that operates on the entire collection rather than discrete segments. It is used within the materialize/5 lifecycle to transform a stream of events into an accumulated state using a reducer pattern.

When using this strategy, the reducer function and trigger receive special metadata identifiers: {:global, :global, :placeholder} for event processing and {:global, :global, name} for triggers. This allows the reducer to distinguish global operations from segment-specific ones.

Flow.Window.Periodic is a windowing strategy that groups events based on a fixed time duration. It uses a timer to trigger the end of a window and the start of a new one.

To use this strategy, you must provide a duration (required) and a reducer_acc function that initializes the accumulator for each new window. The materialize/5 function transforms these parameters into the internal state required by the Flow engine, including the reducer function and the trigger mechanism.