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
