# 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:** 12

<div class="post-metadata">

**Author:** ![Iex.new](https://forum.elixirforum.com/user_avatar/forum.elixirforum.com/iex.new/32/18284_2.png) [@Iex.new](https://forum.elixirforum.com/u/Iex.new)\
**Post date:** [December 27, 2023, 9:41am UTC](https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424/12 "2023-12-27T09:41:08Z")

</div>

> [@Iex.new](#):
>
> `MyXQL.Connection (#PID<0.302.0>) disconnected: ** (DBConnection.ConnectionError)`

Until now I hadn’t managed ☹

Resources I found to help me to try to solve the issue:

- Mysql Documentation: [https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar\_max\_connections](https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_max_connections)

- Ecto configuration:

My configuration:

my.cnf (mysql configuration)

```elixir
max_connections=100000

```

file config/config.exs

```elixir
pool_size: 500,
migration_timestamps: [type: :utc_datetime_usec],
migration_lock: nil,
queue_target: 500

```

into the file that reads the CSV file I set max\_conccurency to 500:

```elixir
def load(path) do
  path
  |> Path.expand()
  |> File.stream!(read_ahead: 524_288)
  |> YourCSV.parse_stream()
  |> Stream.chunk_every(500)
  |> Task.async_stream(fn rows ->
    # do something with a single CSV record.
    # possibly best to also hand off the results to another service?
    # if you want to just process all records and receive a new list
    # then you should just use Stream.map and Enum.to_list at the end.
    Enum.map(rows, fn [name, bar_code, price, currency] ->
       FileWatchExample.Product.create_product(%{name: name, bar_code: bar_code, price: price, currency: currency})
    end)
  end, max_concurrency: 500, on_timeout: :kill_task, ordered: false)
  |> Stream.run()

```

products.ex :

```elixir
... some code

 def create_product(attrs) do
   %FileWatchExample.Product{}
   |> FileWatchExample.Product.changeset(attrs)
   |> Store.insert(timeout: 1000 * 60)
 end

```

---

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