Get Snapshots from Slow-running or Infinite Flows

Slow-running or infinite flows

So far, we’ve used a CSV file as our primary data source, which Elixir processes immediately. To demonstrate how windows and triggers work, we’ll intentionally slow down the file stream events using Process.sleep/1. We’ll also modify our business logic and still group all data by country code. However, this time, we capture progress snapshots of every one thousand events.

Slow down the stream of events

Let’s start by slowing the stream of events:

airports_csv()
  |> File.stream!()
  |> Stream.map(fn event ->
    Process.sleep(Enum.random([0, 0, 0, 1]))
    event
  end)
  |> Flow.from_enumerable()

Since we have tens of thousands of rows in our CSV file, we randomly delay some of the events only for one millisecond. The rest of the events will go through without delay. This means that the flow will still complete (eventually), but it won’t be as quick as before.

We can try running this code version. It should take about a minute to complete.

Define the window variable

Then, we define the ...