Good evening everybody, sorry for delayed update, I had extra hectic working week so put my POC on hold, but today I finally competed first part and got more or less working version.
In short results are just tremendous - I did not expect such a great outcome. And outcome is as following:

In Elixir app:

• each hour, I download ~400 CSV-like files from S3, total size ~70-75Gb. Total # of records to aggregate is ~0.55-0,7 billions. Files are in different size from 0.5Mb to 2,2Gb
• after downloading each files is processed with aggregation function what we discussed here already

the average latency is ~17-18 minutes for the whole work, which is almost 2 times better than I did with Java prototype 1 year ago!!!

The CPU usage is ~50%, the memory usage ~50% which is slightly higher than in Java but totally acceptable (please see screenshot bellow).

I performed 3 test runs for different hours. The test platform is my cloud devbox C5.9XL (36 cores, 25Gb net, 72Gb RAM)

This is first functional step in the POC (and I still need to work a lot) but first results exceed all my expectation!

Thanks to everybody who helped to reach me such results.

kona-elixir-middle1195×678 133 KB

So… a couple of things. First, File.stream! is already splitting the input into lines by default, but you then Stream.map(&String.split(&1, "\n")) again on each line, which does nothing.

If you need to do a large number of updates on a Map, consider switching to ETS. It will perform much better with large amounts of data, since it has constant time access and isn’t garbage collected.

There are some other things I’d try that might help too, like replacing this &((elem(&1, 0) <> "," <> Decimal.to_string(elem(&1, 1))) <> "\n") with fn {key, value} -> key <> "," <> Decimal.to_string(elem(&1, 1)) <> "\n" end.

You’re also building a lot of strings, which in eg the JVM is automatically optimized. The BEAM doesn’t optimize that automatically, but there’s this concept of “iolists” that let you manually optimize it.

I’d suggest taking a look at a similar thread posted before that has a lot of tips Erlang/Elixir string performance - can this be improved?

I also wrote an article about this which isn’t completely up to date but collects most of the improvements from the thread https://blog.jola.dev/elixir-string-processing-optimization and it has examples of eg using ETS instead of Map and using iolists.

Based on my experience optimizing a similar code snippet (spent lots of hours on it, measuring each part).

• Streaming into lines is very slow compared to reading the entire string directly. Writing a stream is also slower than writing a complete string. Keeping unicode support is slower than not.
• Splitting is fast when you can split 1 big thing, but slow when you split many small things.
• Passing a large map across functions is not something that takes time. As long as it stays within the process nothing gets moved.
• Putting values in large map is very slow (depending on your definition of large). The time it takes grows with the size of the map (unlike ETS which is constant time).
• Overhead of stream is small, but noticeable with a large number of items (like this case). Memory is the deciding factor. If you can spare it, don’t use Stream.

Another thing, causing lots of GC by constantly growing the heap. With a similar use case script I was able to speed it up considerably by setting the initial process heap size to a very large number, which avoids growing the heap a bunch of times. GC in general is costly, if you can avoid it, do. But that’s up to your memory limits.

ps come to my talk at CodeElixir LDN or ElixirConf US where I will be talking about this thing exactly

a few performance tricks to use when dealing with Strings in Elixir (or Erlang)

• don’t use string concatenation (<>) use io lists
• don’t use records maps, they are slow, use lists or tuples
• take advantage of the standard library, don’t overthink the solution: in this case Integer.parse is more than enough to convert a string to an integer
• use Erlang’s binary module. For example compiled patterns are usually much faster than just string patterns.

On my Laptop (Lenovo P70, 8 cores, 64GB of RAM) your solutions takes approximately 6 seconds to process a little bit more than 1 million rows, while the following takes around 2.1 seconds

  def aggregate_kona2() do
    IO.puts("Start")
    kona = "./data"

    output_path = kona <> "_elixir_aggregated"
    
    pattern = :binary.compile_pattern(",")

    {time, _} =
      :timer.tc(fn ->
        kona
        |> File.stream!()
        |> Stream.drop(9)
        |> Stream.map(&parse_granular2(&1, pattern))
        |> Enum.reduce(
          %{},
          fn [key, value], acc ->
            Map.update(acc, key, value, &Kernel.+(&1, value))
          end
        )
        |> Stream.map(fn {k, v} -> [k, ",", to_string(v), "\n"] end)
        |> Stream.into(File.stream!(output_path, [:write, :utf8]))
        |> Stream.run()
      end)

    IO.puts("Total: #{time / 1_000_000}s")
  end

  def parse_granular2(record, pattern) do
    [p, pr, cpc, ut, op, _, _, _, _, _, _, value] = record |> String.split(pattern)

    {value, _} = Integer.parse(value)

    [
      [p, ",", pr, ",", cpc, ",", ut, ",", op],
      value
    ]
  end

EDIT

Here’s a version using Flow.

It’s not optimized, I’m sure tweaking the params a bit will yeld much better results.

For around 11 millions rows it take ~17secs (as usual for the BEAM, results scale linearly with the size of the input, for 1 million rows it take 1.55 secs, 1.55 * 11 = 17.05)

  def aggregate_kona_flow() do
    IO.puts("Start")
    kona = "./data12"

    output_path = kona <> "_elixir_aggregated_3"
    {:ok, output_file} = File.open(output_path, [:write, :utf8])

    pattern = :binary.compile_pattern(",")

    part_opts = [stages: System.schedulers_online()]
    read_ahead = 32768
    max_demand = 256

    {time, _} =
      :timer.tc(fn ->
        kona
        |> File.stream!(read_ahead: read_ahead)
        |> Stream.drop(9)
        |> Flow.from_enumerable(max_demand: max_demand)
        |> Flow.map(&parse_granular2(&1, pattern))
        |> Flow.partition(part_opts)
        |> Flow.reduce(fn -> %{} end, fn [key, value], acc ->
          Map.update(acc, key, value, &(&1 + value))
        end)
        |> Flow.each(fn {k, v} ->
          IO.write(output_file, [k, ",", to_string(v), "\n"])
        end)
        |> Flow.run()
      end)

    File.close(output_file)

    IO.puts("Total: #{time / 1_000_000}s")
  end

If you want maximum speed you have to stick with tuples, iolist and, integers. I wouldn’t use Decimal, it is based on structs which is a map, and maps are expansive, not as expensive as they used to be. I would recommend you to use :erts_debug.size/1 to determine how much memory your would use.
:erts_debug.size(%{}) == 4 :erts_debug.size({}) == 1

/edit

Another thing is, use pattern matching, if your data is consistent and you always expect the csv to be the same then you don’t have to split the lines to get your values, you can pick them right out of the string which would give you almost pointer arithmetic.

def getmyvalue(<<"StartRangeMarker=", year::binary-size(4), ?-, month::binary-sizer(2), ?-, day::binary-size(2), ?-, hours::binary-size(2), ?-, min::binary-size(2), ?-, sec::binary-size(2)>>) do
{:ok, year, month, day, hours, min, sec}
end