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

raza_ep

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 1739 6

2023-03-14 07:43:32 UTC

Showing Posts 1 to 6

Show Best Posts
Show All (oldest first)
Show All (newest first)

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