Inserting CSV into postgres

sabri

Hello,

I have found lovely CSV library that I will be using to insert CSV files into postgres.

But there are some issues I am wondering about, as I am fresh in elixir.

Ok, here is the sample code I will be using to insert the CSV into DB:

File.stream!("ignore/customers.csv")
|> CSV.decode
|> Enum.each(fn
        {:ok, [id, nm, csr, sal]} ->
            Customer.changeset(%Customer{},
                %{masterid: id,
                  custname: nm,
                  csrid: String.to_integer(csr),
                  salesid: String.to_integer(sal)})
            |> Repo.insert
         {:error, message} ->
            # Whatever you want to do with invalid rows
    end)

My questions are:

In the Enum.each what is the time interval between each call to the Repo.insert? can I control this to make sure that my DB won’t get over-pumped with queries?
As I need to implement a progress bar in the browser while the CSV being inserted into DB, Can I broadcast a progress message in successful Enum.each to client using channels? for ex: after |> Repo.insert to broadcast message as:

MyApp.Endpoint.broadcast client_topic, “progress”, %{
progress: progress_number
}

15 comments

#ecto #streams

2 7173 15

2017-02-24 06:24:40 UTC

Showing Posts 15 to 6

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

minhajuddin

I did a screencast covering this here https://www.youtube.com/watch?v=YQyKRXCtq4s

Post #15

sabri

Thanks, I’ve finally implemented it!, but I have to admit, that I had to change my mindset to do it
Here is a sample chunk:

[%{first: “Antonio”, seq: “1”}, %{first: “Robert”, seq: “2”},
%{first: “Gordon”, seq: “3”}, %{first: “Tyler”, seq: “4”},
%{first: “Landon”, seq: “5”}, %{first: “Ethan”, seq: “6”},
%{first: “Amy”, seq: “7”}, %{first: “Paul”, seq: “8”},
%{first: “Tom”, seq: “9”}, %{first: “Helen”, seq: “10”}]

As I have used Stream.map in the pipeline:

File.stream!(path)
    |> MyParser.parse_stream
    |> Stream.map(fn n ->
      case n do
        [seq,first,last,email,digit] -> %{seq: seq, first: first}
        _ -> ""
      end
      end)
    |> Stream.chunk(10)
    |> Task.async_stream(&MyApp.CsvsController.chunk_handler_fn/1)
    |> Stream.run

Thanks all for support

Post #14

josevalim

Creator of Elixir

That’s what Stream.chunk(10) does, it builds chunks up to 10 elements. Therefore inside Task.async_stream, you will have list of 10 elements, then you need to use Enum.each or Enum.map and friends to traverse them, process them, and insert them into the database in chunks.

Post #13

sabri

Thanks, all chunks got streamed now, but I could not parse them, here is what I got as a sample chunk from |> Stream.chunk(10)

1AntonioMartinezfemeise@bi.gov3484293762RobertButleramfob@vaim.io7388480513GordonSimonnak@jodcon.org8829569684TylerFlowersid@do.net5789928675LandonRossriojipof@fuju.gov7791899366EthanChavezozeketja@vucocabo.gov7585680127AmyHendersoneruraep@emu.io9750233448PaulWilliamsabmahu@bi.edu7778078969TomHerreraosaik@alaigocuc.com45881961610HelenCurrymuw@pi.io598349524

While NimbleCSV will generate an array for each line in the CSV file, |> Stream.chunk(10) has concatenated the arrays as above, making it impossible to get the data using pattern matching.

So, what’s the advice here?

Post #12

josevalim

Creator of Elixir

If I run the stream as Stream.run will it run all the chunks until the stream is all streamed?

All chunks. But try it.

When will the temp streamed uploaded file get deleted?

Uploaded files are saved to a temporary directory and deleted afterwards, yes.

Post #11

sabri

I do care about results, so I am not using Enum, in-fact, I will be broadcasting results to the client, to implement a progress bar of achievement of the task, as in the top of this thread.

If I run the stream as Stream.run will it run all the chunks until the stream is all streamed? or shall I run it for each chunk?
When will the temp streamed uploaded file get deleted? after finishing the stream?

Post #10

michalmuskala

Task.async_stream returns a lazy stream - it won’t be executed unless you consume it. This can be done with Stream.run if you don’t care for results or any of the functions working on enumerables in an eager way if you do care about return values (most notably from the Enum module).

Post #9

sabri

Ok, here is my humble try with a CSV file that has 100 entry like:

CSV:

seq,first,last,email,digit
1,Antonio,Martinez,femeise@bi.gov,348429376
2,Robert,Butler,amfob@vaim.io,738848051

Code:

NimbleCSV.define(MyParser, separator: ",", escape: "\"")
File.stream!("/Users/samir/Downloads/100.csv")
|> MyParser.parse_stream
|> Stream.chunk(10)
|> Task.async_stream(&MyApp.CsvsController.chunk_handler_fn/1)


def chunk_handler_fn(csv_row) do
    IO.puts csv_row
end

But, chunk_handler_fn was never called, I have attached the simple 100.csv sample file here:

https://s3.eu-central-1.amazonaws.com/test-files-74/100.csv

Is there something I have missed?

Post #8

sabri

Thanks for detailed answer, I will try to do the best thing I can now

Post #7

josevalim

Creator of Elixir

Great answer! From the earlier example, I would definitely use Repo.insert_all. Also the parallel processing can be done with Task.async_stream from Elixir v1.4.

Post #6

Load more posts (5 remaining)