The Pipe Operator, Enum and Stream

Readable data transformations

The pipe operator |> passes the result of one expression as the first argument of the next function call, turning nested calls into a top-to-bottom pipeline: orders |> Enum.filter(& &1.paid) |> Enum.map(& &1.total) |> Enum.sum(). Elixir's library is designed so the data is the first argument, which makes pipes natural. The Enum module works on any enumerable (lists, maps, ranges): map, filter, reject, reduce, sum, group_by, frequencies, sort_by, chunk_every, zip, find, any?, all?, uniq_by and many more. Enum functions are eager, building a full list at each step. Stream functions are lazy: they compose transformations and run only when consumed, ideal for large files (File.stream!), infinite sequences or expensive steps. Comprehensions with for combine generators, filters and an into: option: for %{paid: true, total: t} <- orders, t > 1000, do: t. Use then/2 to pipe into a function where the value is not the first argument, and tap/2 for side effects such as logging inside a pipeline.

Pipelines over orders and a lazy file stream

Eager Enum, lazy Stream and a comprehension.

orders = [
  %{id: "o1", city: "Pune", total: 120_000, paid: true},
  %{id: "o2", city: "Delhi", total: 30_000, paid: false},
  %{id: "o3", city: "Pune", total: 80_000, paid: true}
]

revenue_by_city =
  orders
  |> Enum.filter(& &1.paid)
  |> Enum.group_by(& &1.city, & &1.total)
  |> Map.new(fn {city, totals} -> {city, Enum.sum(totals)} end)
# %{"Pune" => 200000}

city_counts = orders |> Enum.map(& &1.city) |> Enum.frequencies()
# %{"Delhi" => 1, "Pune" => 2}

big_paid = for %{paid: true, total: t, id: id} <- orders, t > 100_000, do: id
# ["o1"]

# lazy: processes a large CSV line by line without loading it all
total_paise =
  File.stream!("orders.csv")
  |> Stream.drop(1)                                   # skip the header
  |> Stream.map(&String.split(String.trim(&1), ","))
  |> Stream.map(fn [_id, _city, total] -> String.to_integer(total) end)
  |> Enum.sum()                                       # consuming runs the stream

IO.inspect({revenue_by_city, city_counts, big_paid, total_paise})

An assembly line

A pipeline is an assembly line: each station takes what arrives, does one job and passes it on. Enum runs every item through one station before moving to the next; Stream sends each item down the whole line one at a time.

त्वरित जाँच: Where does |> place the value from the left side?

• As the first argument of the function on the right
• As the last argument
• In a global variable
• It replaces all arguments