stevensonmt

Stream MUCH slower than enum processing modest CSV file

I’m working through the book Concurrent Data Processing in Elixir and the example in chapter four involves reading a smallish CSV file and filtering out some rows. The naive implementation uses File.read! and NimbleCSV.parse_string and does the filtering with the Enum module functions.

A slightly better example uses File.stream! and NimbleCSV.parse_stream and does the filtering with the Stream module functions before finally collecting to a list with Enum.to_list.

The first version runs in about 4 seconds on my machine. The second takes over two minutes! In the book they reported a 5-fold improvement from ~3 seconds to 600ms. What could explain the discrepancy?

Original:

defmodule Airports do
  alias NimbleCSV.RFC4180, as: CSV

  def airports_csv() do
    Application.app_dir(:airports, "/priv/airports.csv")
  end

  def open_airports() do
    airports_csv()
    |> File.read!()
    |> CSV.parse_string()
    |> Enum.map(fn row ->
      %{
        id: Enum.at(row, 0),
        type: Enum.at(row, 2),
        name: Enum.at(row, 3),
        country: Enum.at(row, 8)
      }
    end)
    |> Enum.reject(&(&1.type == "closed"))
  end
end

Stream version:

  def open_airports() do
    airports_csv()
    |> File.stream!()
    |> CSV.parse_stream()
    |> Stream.map(fn row ->
      %{
        id: :binary.copy(Enum.at(row, 0)),
        type: :binary.copy(Enum.at(row, 2)),
        name: :binary.copy(Enum.at(row, 3)),
        country: :binary.copy(Enum.at(row, 8))
      }
    end)
    |> Stream.reject(&(&1.type == "closed"))
    |> Enum.to_list()
  end

Even weirder, the book then introduces the Flow library to replace the Stream functions and reports a two fold slow down, but I get a 60 fold speed up!

  def open_airports() do
    airports_csv()
    |> File.stream!()
    |> CSV.parse_stream()
    |> Flow.from_enumerable()
    |> Flow.map(fn row ->
      %{
        id: :binary.copy(Enum.at(row, 0)),
        type: :binary.copy(Enum.at(row, 2)),
        name: :binary.copy(Enum.at(row, 3)),
        country: :binary.copy(Enum.at(row, 8))
      }
    end)
    |> Flow.reject(&(&1.type == "closed"))
    |> Enum.to_list()
  end

Finally, fixing the presumed bottleneck to allow flow to take advantage of concurrency by moving the CSV parsing into the Flow.map call is supposed to lead to significant speed up but in my case is essentially the same running time.

  def open_airports() do
    airports_csv()
    |> File.stream!()
    |> Flow.from_enumerable()
    |> Flow.map(fn row ->
      [row] = CSV.parse_string(row, skip_headers: false)

      %{
        id: Enum.at(row, 0),
        type: Enum.at(row, 2),
        name: Enum.at(row, 3),
        country: Enum.at(row, 8)
      }
    end)
    |> Flow.reject(&(&1.type == "closed"))
    |> Enum.to_list()
  end

I really am at a loss to explain why I’m having such different results from the book.

8 comments

#troubleshooting

12 1054 8

2025-08-16 15:04:38 UTC

Most Liked

LostKobrakai

Is this the same code as well as the same data?

Generally moving from enum to streams trades speed for less memory need. Things are not meant to become faster (but rather become slower) in favor of not needing to load everything into memory at once and being able to process data, which doesn’t fit into memory comfortably or even at all.

If your data fits into memory Enum functions are expected to be faster.

Similarly moving to a multi process architecture adds overhead in process communication (there’s no reduction) in favor of being able to use multiple cpu cores at the same time.

If your data doesn’t even max out a single core for processing it’ll be faster in a single process than with multiple ones.

Especially the last one is a bit simplified as there’s also things like io involved, but I hope you get the idea.

Post #2

LostKobrakai

While this is a nice hint for the next person, testing performance in iex is not a great idea anyways. There are lots of behaviours different in iex to a production environment that it cannot reasonably serve as a proxy for how performance will look elsewhere.

Post #8

stevensonmt

It’s real world data, so the file is probably not exactly the same as the one used by the authors, but it is from the same source and should be similar in size and structure. The code I’m using is the same as from the book, yes.

Thanks for explaining Enum vs Stream broadly. As I understand it both inherently and from your explanation, the performance results I got probably make some sense but I still don’t understand why my results diverge so drastically from the authors’ results.

Just to reinforce @LostKobrakai’s points above, the chapter concludes with the same basic advice about when using something like Flow for concurrency might be a bad idea:

While processes are lightweight, they are still an overhead when dealing with
small tasks, which can be processed faster synchronously.

Post #3

SebAlbert

I found this old topic when I was searching for the exact same observation I had today while working through that section. Also on my machine the time it took went up from 3.5 seconds with the File.read! |> CSV.parse_string variant to 120-130 seconds with File.stream! |> CSV.parse_stream.

However, while reading on, I stumbled across the hint towards the “read_ahead” option in File.stream!/2 and wanted to revisit the Stream implementation to see if that option had any impact. But before I even tried that option, to my surprise I suddenly had the Stream variant run in approximately 1 second.

I played around a bit and found the following:

If I first run the File.read variant without :binary.copy, and afterwards in the same iex session (after recompile()) the Stream variant, it takes 2 minutes.
If I start with a fresh iex session, the Stream variant takes about 1 second.
If I first run the File.read variant with :binary.copy, and afterwards in the same iex session (after recompile()) the Stream variant, it also only takes 1 second.

So there might be something hanging on to the large string(s) in memory in the interacive session, I suppose.

Post #7