Elixir language performance tuning for 1 quadrillion records per month

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 Elixir String Processing Optimization | jola.dev and it has examples of eg using ETS instead of Map and using iolists.

Thank you. Let me revisit impl with suggestions you gave and I’ll be back with (hopefully) new results.

Sorry could you help to rewrite a reduce function using ETS from the example I have? I desperately spent 3 hours doing that with no result and getting

** (ArgumentError) argument error

This is what I am trying

table = :ets.new(:records, [])

  kona
  |> File.stream!
  |> Stream.drop(9)
  |> Stream.map(&parse_granular(&1))
  |> Enum.each(fn record -> :ets.update_counter(table, record.record_key, {2, record.record_value}, {record.record_key, record.record_value}) end)

Erlang doc is not clear at all.

Thank you.

 update_counter(...)

The function fails with reason badarg in the following situations:

The table type is not set or ordered_set …The element to update is also the key. …

Here’s an example:

defmodule A do

  require Record
  Record.defrecord(:user, record_key: 0, record_value: "", visits: 0)
  #default record =>  {:user, 0, "", 0}

  def go do
    :ets.new(:mytable, [:named_table, {:keypos, 2}]) #tuple indexes start at 1 in erlang
                                                     #keypos is the position in the tuple that serves 
                                                     #as the index in the table, which in this case 
                                                     #is record_key
    #defaults for :ets.new():                                                             
    #[set, protected, {keypos,1}, {heir,none}, 
    #{write_concurrency,false}, {read_concurrency,false}].

    records = [
      user(record_key: 1, record_value: "hello"),
      user(record_key: 2, record_value: "goodbye", visits: 1),
      user(record_key: 3, record_value: "g'day", visits: 2)
    ]

    Enum.each(records, fn record -> 
                          :ets.insert(:mytable, record)
                       end)
    IO.inspect :ets.tab2list(:mytable)

    Enum.each(records, fn user(record_key: key) -> 
                          :ets.update_counter(:mytable, key, {4, 1})  #{tuple pos, increment}
                       end)
    IO.inspect :ets.tab2list(:mytable)
    :ok
  end

end

In iex:

~/elixir_programs$ iex a.ex
Erlang/OTP 20 [erts-9.3] [source] [64-bit] [smp:4:4] [ds:4:4:10] [async-threads:10] [hipe] [kernel-poll:false]
Interactive Elixir (1.8.2) - press Ctrl+C to exit (type h() ENTER for help)

iex(1)> A.go
[{:user, 1, "hello", 0}, {:user, 3, "g'day", 2}, {:user, 2, "goodbye", 1}]
[{:user, 1, "hello", 1}, {:user, 3, "g'day", 3}, {:user, 2, "goodbye", 2}]
:ok

iex(2)>

You’re using update_counter, which only supports updating integers, but your value is some struct. It is in the documentation, but I also think erlang documentation is hard to read.

You’ll need a combination of lookup and insert. Something like

def increment_item(table, key, value) do
  initial = case :ets.lookup(table, key) do
    [{_, value}] -> value
    [] -> Decimal.new(0)
  end

  :ets.insert(table, {key, initial + value})
end

Note that I have no idea what Decimal is, so I don’t know how to create them or how to add them together, so I just used +, you’ll have to replace that. I haven’t tested the code.

Thank Jola, I spent some more time an came up with similar solution myself and I have huge (but not the best) result

Here my implementation

defmodule Kona do

  def aggregate_kona() do

  IO.puts("Start")
  kona =  "/home/temp/1/real_kona"

  output_path = kona <> "_elixir_aggregated"

  start = :os.system_time(:millisecond)
  table = :ets.new(:records, [:set, :protected])

  kona
  |> File.stream!
  |> Stream.drop(9)
  |> Stream.map(&parse_granular(&1))
  |> Enum.each(fn record -> update(table, record) end)

  :ets.match_object(table, {:"$0", :"$1"})
  |> Stream.map(fn {key, value} -> key <> "," <> Decimal.to_string(value) <> "\n" end)
  |> Stream.into(File.stream!(output_path, [:write, :utf8]))
  |> Stream.run()

  stop = :os.system_time(:millisecond)
  IO.puts("Total: #{stop - start}")

  end

  def update(ets_table, record) do
    entry = :ets.lookup(ets_table, record.record_key)
    :ets.insert(ets_table, {record.record_key, Decimal.add(record.record_value, empty?(entry))})
  end

  def empty?([]) do
    Decimal.new(0)
  end

  def empty?(list) do
    list |> hd |> elem(1)
  end

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

    %{record_key: p <> "," <> pr <> "," <> cpc <> "," <> ut <> "," <> op, record_value: Decimal.new((value |> String.replace("\n", "")))}

  end
  

end

with that I can now DO aggregate the big (2Gb) file with latency ~135 sec, but it is still ~8x slower than my java version. Could you please take one more look to see if I can do any better?

Thank you for you time and help!

with your version of reduction function got 15 sec boost more and final latency now is ~120 sec. But my question for more review still valid. Thank you

I’m not that sure, but…
I’m not surprised that this version is slower, because elixir/erlang is slower than java in general. To outperform java you need to use currency, which is not used in this case. Stream just helps with memory usage AFAIK.

You probably should look into Flow — Flow v1.2.4, which will parallelize your computations.

Disclaimer: I may be talking bs…

That’s not entirely true, stricto senso: streams allow composing operations so the collection is iterated over only once. In other words, if you want to apply multiple transformations to a collection of items, if you use Map the collection will be traversed once per transformation, but if you use Stream all the transformations will be “collected” into a single operation and only then will the collection be traversed (and only once).

Yes, and this is why it helps with memory consumption, we have to pay this though. Streams have an additional computational overhead and often slower than their eager counterparts.

But as we have an arbitrary large input file, it sounds as if the price is worth to be paid.

Anyway, I doubt as well, that the BEAM version will be able to be as performant as the Java version.

The BEAM version involves a lot of copying data back and forth between the process and the ETS, which in the Java Version is probably mutated in place.

Still, I’d like to give some more suggestions:

1. Check the other kinds of ETS tables. Some of them are slower than others, and as far as I remember set is somewhat costly on insert/update, as it always has to scan the table for the key, but maybe I’m missremembering things from the talk (it was briefly mentioned in the talk about how phoenix got to 2M active connections)
2. The parse function could probably be improved by not splitting and reconstructing a big portion of the string but instead found the position of the xth comma and the last comma, and use String.slice/2/3 to get the string-id and value.
3. don’t create a map in the parser that you then will restructure into a tuple all over the place. Make and keep it as a tuple.
4. instead of doing a match all query to the ETS, you could check if :ets.tab2list is more performant.