I built an Erlang implementation that does 1B lines in 60 seconds on my slow laptop.

To improve it further, maybe one could use the counters module. So much to do, so little time…

That post probably offers better insight than I can.

There’s also this blog post which offers the following explanation:

An ETS based approach
Other times, if the Agent doesn’t cut it for you, you might something faster. In these cases ETS might be a good option. The good thing about ETS is that it will always be faster because it doesn’t go through the Erlang Scheduler, furthermore it also supports concurrent reads and writes, which the Agent does not. However, it’s a bit more limited when you want to do atomic operations. Overall it’s very well suited for a simple shared key/value store, but if it’s better suited or not for your specific problem, that’s up to you.

Finally, the approach that immediately came to my mind was not using a cache at all but using Task.asyn_stream as described in this blog post

Another implementation outside of the standard library if you’re interested.

There’s probably some small gains to be had by changing the worker_pool to a map rather than list so that you can avoid Enum.at here

Agent.cast(Enum.at(worker_pool, rem(index, @pool_size)), Brc, :process_lines, [job])

It has to iterate over the worker_pool list each time to find the agent id. Since the worker pool is small each individual call is going to be very fast, but you are calling it 100_000 times. On my machine the average of 100 runs of 100_000 calls to Enum.at on a list of 8 items versus calling Map.get on a map of 8 items keyed by index was a difference of 17631ms. Whether that is significant is for you to decide. Probably the Agent issue is more impactful.

My rough little benchmark:

list = 1..8 |> Enum.to_list
map = list |> Enum.with_index() |> Map.new(fn {v, k} -> {k, v} end)
fun = fn n -> 1..n |> Enum.map(fn i -> Enum.at(list, rem(i, 8)) end) end
fun2 = fn n -> 1..n |> Enum.each(fn i -> Map.get(map, rem(i,8)) end) end
1..100 |> Enum.map(fn _ -> :timer.tc(fn -> fun.(100_000) end) end) |> Enum.map(&elem(&1, 0)) |> Enum.sum() |> div(100)
# 157815
1..100 |> Enum.map(fn _ -> :timer.tc(fn -> fun2.(100_000) end) end) |> Enum.map(&elem(&1, 0)) |> Enum.sum() |> div(100)
# 140184

I think you may have changed some other parts of the code as well, looking at the github repo. So you might have minimized the impact the change could have. I just re-ran the benchmark on my machine, using an ETS implementation and got similar results as I posted above:

:ets.new(:sample, [:named_table, :public])
map |> Enum.each(fn {k, v} -> :ets.insert(:sample, {k, v}) end)
fun3 = fn n -> 1..n |> Enum.each(fn i -> :ets.lookup(:sample, rem(i,8)) end) end
1..100 |> Enum.map(fn _ -> :timer.tc(fn -> fun3.(100_000) end) end) |> Enum.map(&elem(&1, 0)) |> Enum.sum() |> div(100)
# 131589
1..100 |> Enum.map(fn _ -> :timer.tc(fn -> fun2.(100_000) end) end) |> Enum.map(&elem(&1, 0)) |> Enum.sum() |> div(100)
# 139659
1..100 |> Enum.map(fn _ -> :timer.tc(fn -> fun.(100_000) end) end) |> Enum.map(&elem(&1, 0)) |> Enum.sum() |> div(100)
# 152674

Also be aware that while ETS lookups are fast (O(1) as you say) and process communication is very efficient in the BEAM, making process calls can be slower than accessing a data structure in the same process. ETS will generally be a better choice for very large collections and for structures needing to be accessed from multiple concurrent processes. In this case your worker_pool is never going to be that large. Whether sharing a single ETS across multiple processes is more efficient than creating the worker_pool map for each process I can’t say but probably it is given how many processes you could end up spawning to access it. This thread has a good discussion of the tradeoffs between maps and ETS tables.

Rather than focusing on this tiny worker_pool data structure, I think you should look at processing all the lines directly to an ETS table rather than creating a bunch of maps stored in Agents that you have to merge later. Create Tasks that process lines to the ETS table then at the end pull the table data for your final output. If you set up your ETS table as an ordered_set you can probably avoid the step where you’re having to sort N items where N is the number of cities in the data set. You might event look at DETS to avoid running out of memory given the large data set.