# Data processing in Parallel (Hackathon project)

**URL:** <https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424>\
**Category:** Questions\
**Tags:** phoenix, flow, broadway\
**Created:** [December 17, 2023, 6:12pm UTC](https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424 "2023-12-17T18:12:57Z")\
**Posts on this page:** 1\
**Showing post:** 29

<div class="post-metadata">

**Author:** ![dimitarvp](https://forum.elixirforum.com/user_avatar/forum.elixirforum.com/dimitarvp/32/38664_2.png) [@dimitarvp](https://forum.elixirforum.com/u/dimitarvp)\
**Post date:** [January 3, 2024, 9:05am UTC](https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424/29 "2024-01-03T09:05:39Z")

</div>

Now that you posted the entire code, here’s exactly what’s going on.

You are spawning 18 parallel processes and each of them is trying to insert 500 records _one by one_ which means 9000 DB transactions. Unless you configure your pool with 9000 connections then an incomplete ingestion is to be expected.

The idea of balancing out your Repo’s pool size and the streaming processing’s `max_concurrency` parameter is to avoid exactly what your code is doing: never go above the Repo’s pool size.

But your code is doing it, in fact exceeding Repo’s capacity by 500x.

Change your code to do this:

```elixir
defmodule IngestCSV do
  alias NimbleCSV.RFC4180, as: YourCSV

  require Logger

  NimbleCSV.define(YourCSV, separator: ";")

  def load(path) do
    path
    |> Path.expand()
    |> File.stream!(read_ahead: 524_288)
    |> YourCSV.parse_stream()
    # Remove `Stream.chunk_every`
    |> Task.async_stream(fn [name, bar_code, price, currency] ->
       # Remove `Enum.map`
       FileWatchExample.Product.create_product(%{name: name, bar_code: bar_code, price: price, currency: currency})
    end, max_concurrency: 18, on_timeout: :kill_task, ordered: false)
    |> Stream.run()
  end
end

```

Now your DB pool’s size is 20 and you will never have more than 18 records being inserted in parallel, thus you should never have dropped ingestions.

If that really works then you can proceed to bump up the Repo’s pool size to e.g. 100 and put `max_concurrency` at 95 - 98 and that should work fine and accelerate your workload.

I believe you are being tripped up by `Task.async_stream`’s semantics. The function that is passed to it is going to be working in 18 _separate parallel independent processes_. BUT, having `Enum.map` inside of it is still serial i.e. _not_ parallelized. So you do have 18 parallel independent tasks each trying to open 500 DB transactions (inserts) one after another. As you have found that that mostly works but not always – because it’s a very wrongly written parallel code.

---

_[View the full topic](https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424)._