raza_ep

Best way to read data in the most performant way from multiple large CSV files?

My objective is to read data (in the most performant approach) from multiple large CSV files, make minimal transformations and eventually write to output files. I want to ensure I am leveraging all the compute on my machine.

I am using the following deps:

      {:flow, "~> 1.2.3"},
      {:nimble_csv, "~> 1.2.0"},
      {:parallel_stream, "~> 1.1.0"}

Let me know your thoughts if this code is performant.

defmodule Sample do
  alias NimbleCSV.RFC4180, as: CSV


  def process_data(datafile) do
    datafile
    |> File.stream!()
    |> Flow.from_enumerable()
    |> Flow.map(fn row ->
      [row] = CSV.parse_string(row, skip_headers: false)
      %{
        id: :binary.copy(Enum.at(row, 0)),
        name: :binary.copy(Enum.at(row, 2)),
        place: :binary.copy(Enum.at(row, 4))
      }
    end)
    # |> Enum.to_list()
    |> Flow.run()
  end

  def read_files() do
    Path.wildcard("data/*.csv")
  end

  def init() do
    read_files()
    |> ParallelStream.map(fn file ->
      process_data(file)
    end)
    |>Enum.into([])
  end

end

Sample.init() invokes the file(s) processing.

Thanks,
Raza

6 comments

#flow #learning-elixir #csv #optimization

0 1583 6

2023-03-14 07:43:32 UTC

First 6 of 6 Posts

al2o3cr

One note unrelated to performance - File.stream! returns lines by default, but a CSV row may span multiple lines if it contains an embedded \n. For instance, this is a valid CSV with one row and three columns:

foo,"bar
baz",wat

parse_stream has additional machinery to handle escapes (like the " before bar on line 1 above) that cross line boundaries, but parse_string assumes the given string contains the whole CSV file.

Post #1

Schultzer

I would recommend you benchmark your code: Readme — Benchee v1.5.1.

And I’m corious as to why you use :binary.copy like that?

Post #2

BradS2S

raza_ep:

y objective is to read data (in the most performant approach) from multiple large CSV files, make minimal transformations and eventually write to output files. I want to ensure I am leveraging all the compute on my machine.

I am using the following deps:
      {:flow, "~> 1.2.3"},
      {:nimble_csv, "~> 1.2.0"},
      {:parallel_stream, "~> 1.1.0"}
Let me know your thoughts if this code is performant.
defmodule Sample do
  alias NimbleCSV.RFC4180, as: CSV


  def process_data(datafile) do
    datafile
    |> File.stream!()
    |> Flow.from_enumerable()
    |> Flow.map(fn row ->
      [row] = CSV.parse_string(row, skip_headers: false)
      %{
        id: :binary.copy(Enum.at(row, 0)),
        name: :binary.copy(Enum.at(row, 2)),
        place: :binary.copy(Enum.at(row, 4))
      }
    end)
    # |> Enum.to_list()
    |> Flow.run()
  end

  def read_files() do
    Path.wildcard("data/*.csv")
  end

  def init() do
    read_files()
    |> ParallelStream.map(fn file ->
      process_data(file)
    end)
    |>Enum.into([])
  end

end
Sample.init() invokes the file(s) processing.

Thanks,
Raza

Looks like File.Read is faster than File.Stream: Surprising behavior of File.stream vs File.read

Post #3