Stream CSV file from a remote zip on S3

Hello,

I don’t know if it will help, as I’m not using Unzip, but StreamGzip, in combination with NimbleCSV, for this purpose. But maybe it will give you some hint?

I have the following function that returns a stream for an object downloaded from S3:

  defp get_object_stream(object) do
    {:ok, io_pid} = StringIO.open(object)

    io_pid
    |> IO.binstream(4096)
    |> StreamGzip.gunzip()
    |> NimbleCSV.RFC4180.to_line_stream()
  end

In my case, the trick was to use to_line_stream.

I then can use this stream like this:

object
|> get_object_stream()
|> NimbleCSV.RFC4180.parse_stream()

As you can see, I’m not streaming directly from S3 as I first download the object in memory, but if you have something that’s already able to stream from S3, you would just have to replace the part that constructs the stream from the in-memory string with your stream from S3.

Echoing what @ahamez has already mentioned, the issue seems to be that CSV.decode expects stream of lines. But Unzip.file_stream! returns stream of blobs. You can convert stream of blobs to stream of lines yourself, or you can use NimbleCSV as already mentioned.

Unzip.file_stream!(unzip, file_name)
|> NimbleCSV.RFC4180.to_line_stream()
|> NimbleCSV.RFC4180.parse_stream()

Thanks for you replies.

Unfortunately, applying to_line_stream and parse_stream yields an error:

stream 
|> NimbleCSV.RFC4180.to_line_stream() 
|> NimbleCSV.RFC4180.parse_stream() 
|> Stream.run()

** (ArgumentError) errors were found at the given arguments:

  * 1st argument: not a bitstring
    :erlang.bit_size(["\uFEFFservice_id,monday,tuesday,wednesday,thursday,friday,saturday,sunday,start_date,end_date\r\nE1-5-1-127,1,1,1,1,1,1,1,20220115,20220115\r\nH2-0-1-1,1,0,0,0,0,0,0,20220103,20220415\r\nH2-0 (...)
    (nimble_csv 1.2.0) lib/nimble_csv.ex:393: NimbleCSV.RFC4180.to_line_stream_chunk_fun/3
    (elixir 1.12.2) lib/stream.ex:264: anonymous fn/4 in Stream.chunk_while_fun/2
    (elixir 1.12.2) lib/enum.ex:4280: Enumerable.List.reduce/3
    (elixir 1.12.2) lib/stream.ex:931: Stream.do_list_transform/7
    (elixir 1.12.2) lib/stream.ex:1719: Enumerable.Stream.do_each/4
    (elixir 1.12.2) lib/stream.ex:880: Stream.do_transform/5
    (elixir 1.12.2) lib/stream.ex:649: Stream.run/1

What I don’t understand, is the return structure of Unzip.file_stream!. I would expect the stream to return some data, chunk by chunk. If I do Unzip.file_stream! |> Enum.to_list(), I thought I would get something like ["some binary data", "some other binary data", "..."]. Instead I get a nested list of data, that looks like this :

[
 [
  [
   ["some data"], 
   "some other data"], 
   "data again"`]
]

That’s why to_line_stream fails. It expects an enumerable of binaries.

Ok thanks, I needed to adapt the stream coming from Unzip like so :

Unzip.file_stream!(unzip, file_name) 
|> Stream.map(fn c -> List.flatten(c) |> Enum.join("") end) 
|> NimbleCSV.RFC4180.to_line_stream() 
|> NimbleCSV.RFC4180.parse_stream() 
|> Enum.to_list() 

Thank you all for your kind assistance and for giving me good tips!

List.flatten(c) |> Enum.join("") would probably better replaced with IO.iodata_to_binary/1

Nice, thanks @LostKobrakai

I wrote a blog post on the subject, if than can be useful to someone.