I spent a fair amount of time doing large-scale data processing in Elixir. Once you have things loaded up into a queue of some sort, then you have many more tools available for handling failures. But that first pass over the big data file is hard to avoid, and I’m not aware of any tools in any language which fully “solve” for the problem of keeping track of your progress while doing that initial processing of the file.

I didn’t find a silver bullet for recovering seamlessly from crashes, but here are 2 methods I found to help with the chore of getting data from the files and into the queue:

1. Use Bash’s split utility to separate a larger file into smaller files. This helps you divide and conquer, so if things do really go off the rails, you can at least keep track of which files have been successfully processed. One pattern I used grouped items together and relied on Ecto’s insert_all/2 function – this was much more efficient than performing individual database operations, for example:

# Split will create files with an `x` prefix, e.g. `xaa`, `xab`, etc.
input_files = ~w(
/foo/xaa
/foo/xab
/foo/xac
# ... etc...
)

Task.Supervisor.start_link(name: TmpTaskSupervisor)
Task.Supervisor.async_stream_nolink(
  TmpTaskSupervisor,
  input_files,
  fn input_file ->
    input_file
    |> File.stream!()
    |> Stream.chunk_every(1000)
    |> Stream.each(fn chunk ->
      rows = chunk
        |> Enum.map(fn line ->
          %{payload: String.trim(line), foo: "bar", etc: "etc"}
        end)
      MyApp.Repo.insert_all(MyApp.Something, rows)
    end)
    |> Stream.run()

  end,
  timeout: 86400_000,
  max_concurrency: 50
)
|> Enum.to_list()

2. I had some success in some cases using Stream.with_index/2 to help me keep track of which line I was processing – e.g. I could write this number to a file (but in some circumstances, e.g. on AWS, these file operations became a bottleneck). If I needed to recover after a failure, I could pass the number of the last successful line processed to Stream.drop/2 and use that to skip past rows that had already been processed. This can still take a while on a long file, but it’s much faster than re-parsing the whole file.

Relatedly: I’ve been doing some benchmarking of Elixir (e.g. with Python), and one of the tasks involved reading over a large CSV. There are a handful of different patterns I tried, e.g. ex_vs_py/lib/eds/vet_files/control.ex at main · fireproofsocks/ex_vs_py · GitHub My hot take on this was that Python was quite performant for these types of quick one-off tasks – I think any tool is fair game for you to “prime the pump” and get your data out of files and into a queue so you can benefit from the supervision tools available to Elixir as it handles the long-running process.

Hi @artem!

I think the architecture can be quite similar to the Java one. You can use either NimbleCSV or Explorer packages for parsing CSV. For processing later on, you can use Oban or something like Broadway RabbitMQ.

It is quite common but as others have said, your scenario is quite specific so there’s no custom library just for that.

Your task sounds like you would read every CSV record and put them in a DB and mark them with “not processed yet”, and then have a consumer that will dispatch them to processing agents. Very easy and trivial in Elixir, especially to make use of all CPU cores and maximize throughput. I’ve done such tasks in Elixir, Java, Golang, Rust.

Look at Oban and Broadway, they have what you need.