Work with Lazy Streams

Our task is to select active users and return their names. A list holds the source records, and a stream describes how to process them. We will filter, transform, limit, and finally collect results. The important idea is that building a stream pipeline and consuming it are different moments. We will make that visible with a diagnostic step, then discuss why a stream should be treated as a single use computation.

let users = [
    { name: "Ada", active: true },
    { name: "Linus", active: false },
    { name: "Grace", active: true }
]

Follow along with the downloadable example.

Stream converts the list into a lazy source. Where retains records whose active field is true. Map turns each retained record into its name, so the element type changes from a map to a string. Take caps the number of resulting names. At this point we have described the work; we have not yet requested the complete list of answers. Keep the source list if you expect to build another independent computation later.

let pending = users
    |> stream
    |> where { user -> user.active }
    |> map { user -> user.name }
    |> take(2)

Follow along with the downloadable example.

Collect is the terminal operation in this example. It requests the stream elements and materializes them in a list. The output contains Ada and Grace because Linus failed the predicate. We can print names again because it is now a reusable list. That convenience has a cost: collection stores the results. For large or unbounded input, decide whether you really need a complete list or should process a bounded amount instead.

let names = pending |> collect
print(names)
# ["Ada", "Grace"]

Follow along with the downloadable example.

Inspect provides a visible diagnostic without changing each element. Run this file and watch the ordering: before collect appears first, then the inspected numbers, then the collected list. Creating the pipeline did not eagerly print every source item. The consumer pulled the work through the stages. Use this technique while learning, but keep production transformations focused on data and remove noisy diagnostics when they no longer help explain the behavior.

let pending = [1, 2, 3]
    |> stream
    |> inspect { n -> print(n) }
    |> take(2)
print("before collect")
let result = pending |> collect
print(result)

Follow along with the downloadable example.

Filtering before take asks for the first two matching values. Taking two before filtering would only inspect the first two source values, which do not match this predicate, so the result would be empty. Both pipelines are meaningful, but they answer different questions. State your question in ordinary language before choosing the order. A limit usually describes either how much source input to inspect or how many accepted results to keep; those are different boundaries.

[1, 2, 3, 4, 5]
    |> stream
    |> where { n -> n > 2 }
    |> take(2)
    |> collect
    |> print
# [3, 4]

Follow along with the downloadable example.

Flat map is useful when each input element produces a smaller stream. Here each batch is a list, so the closure converts that batch into a stream. Flat map concatenates those child streams lazily, and collect produces one list containing one, two, three, four. Ordinary map would keep one result per batch instead of flattening their elements. Notice the explicit stream conversion inside the closure: the operator expects a child stream, not an arbitrary list.

let batches = [[1, 2], [3, 4]]
batches
    |> stream
    |> flat_map { batch -> batch |> stream }
    |> collect
    |> print

Follow along with the downloadable example.

Treat a stream as single use. When you need to repeat a computation, build a fresh stream from the reusable source, as shown here. Alternatively, collect once and reuse the resulting list if the result is small enough. Do not rely on a second terminal operation to replay the original source. The exact diagnostic for a consumed stream may depend on the operator, so make ownership and consumption clear in the structure of your script.

let source = [1, 2, 3]
let first_run = source |> stream |> take(2) |> collect
let second_run = source |> stream |> take(2) |> collect
print(first_run)
print(second_run)

Follow along with the downloadable example.

Collect needs the stream to finish. Later, file watchers and timer sources may continue indefinitely, so collecting them without a bound is not a way to obtain a quick sample. This finite range example shows the intended pattern safely: add a meaningful limit before materialization. Take also closes upstream early after reaching its count. Choose the bound from the task rather than copying an arbitrary number into every pipeline.

# Prefer a bounded teaching example:
0..100
    |> stream
    |> take(5)
    |> collect
    |> print
# [0, 1, 2, 3, 4]

Follow along with the downloadable example.

For practice, add another active record and increase the result limit to three. Predict the names, run the script, and compare. Then move take before where and explain any difference using the distinction between source items and accepted results. You can now construct a source, transform it lazily, and choose when to consume it. Next we will apply these habits to paths and files, where the data comes from your own working directory.

# Add an active user named Katherine.
# Change take(2) to take(3).
# Predict the names before running.
# Then move take before where and compare.

Follow along with the downloadable example.