hubertlepicki

hubertlepicki

I need to process a bunch of large CSV files on regular basis. They can be multiple gigabytes in size.

Ideally, I would stream the file over HTTP (I can do it with several HTTP clients it seems), and turn it into Elixir Stream and then consume by the Elixir code the same way I consume any other Stream of bytes. I am doing that with Finch at the moment using stream function: Finch — Finch v0.23.0

The problem is that the processing can be interrupted by HTTP connection errors, the server I am streaming the data from seems to have a connection time limit of 30s and then we’re out of luck and have to start over.

I need to re-connect to the same endpoint, then make a request that specifies HTTP “Range” header to start from the point we stopped, and carry on.

I can implement it myself but I suspect someone may have done it already and I just can’t find a library like that? Anyone?

Showing Posts 1 to 4

trisolaran

trisolaran

Interesting question. I don’t know of any library, but I’ve recently built something similar - in my case the source of the data is a rest API providing data about invoices and invoice lines. My code abstracts away the API’s details and creates an Elixir stream that can be consumed. In my case the stream generates a series of invoice lines (each one is a maps), whereas in your case it’s generating bytes.

Also, if I understand your use case correctly, each stream generates a single HTTP request. In my case however, it takes a large number of calls to different endpoints to extract all the data that I need. Similar to your case, sometimes some calls may fail (for different reasons) and I want the stream to protect the consumer from this. Therefore, I built a retry mechanism in the stream that retries failed requests a number of times. As long as the maximum number of retries is not exceed, the consume won’t notice anything and will get its data in the end.

In my case the retry mechanism was simple - just retry the failed request. I guess your case is more complicated because each request fetches a large amount of data and you don’t want to throw away what you’ve already downloaded if the request fails in the middle.

I’m sure you already have good ideas on how to implement the error recovery in case you haven’t found a library.

From my part, a possible idea could be to split the large download into smaller requests that use Range requests from the start, so that none of them exceeds the connection timeout limit.

shamanime

shamanime

@hubertlepicki care to share what you ended up with?

A friend is having issues with a very similar case.

derpycoder

derpycoder

If retries on API failure works, then maybe the following library can help:

https://github.com/wojtekmach/req

I feel like it’s responsibility of providers to have resumability in mind when designing services. Like chunks or Torrents for download, & multipart or Tus protocol for upload.

If the source is well designed, then perhaps the following library can help:

https://github.com/dashbitco/broadway

kodepett

kodepett

Hi, you can take a look at this article from [Poeticoding] - Elixir stream and large http response processing (Elixir Stream and large HTTP responses: processing text). Hope it help.

— All posts loaded —

Where Next? Top

Trending in Questions Top

RSP87
I’m working on a project that simulates the bumbl example in the programming phoenix book. It acts almost like an email client. We have a...
New
RemyXRenard
I’m seeing that a list inside a Kino.DataTable will be interpreted as a charlist, even if the Kino.configure() is set to charlists: :as_l...
New
velrest
So my question is quite simple and i have found no conclusive answer on forum, google or AI. Should we use :erlang.float for Integer to ...
New
nseaSeb
Hello, I know there is an approach for handling lists that allows for optimized traversal, but I can’t recall the specific method (somet...
New
samoloth
Hi, I’ve just set up an application with ash_authentication. There is only magic link strategy for now, so there is no confirmation add o...
New
FlyingNoodle
If a change or preparation module uses Ash.Changeset.get_argument/2 or Ash.Query.get_argument/2 (or any of the other get_argument functio...
New
ryanwinchester
apply_graft/2 doesn’t rewrite an add_many sub-workflow’s deps on an add step. Grafted jobs cancel with “upstream job was deleted” Version...
New

Other Trending Topics Top

mudasobwa
I am happy to introduce the very α version of the new programming language compiled to BEAM. Welcome Cure. It has literally three kille...
New
marciok
Hi there! We created Gust: A task orchestrator inspired by Airflow. For those who have never heard about Aiflow, it’s a Python-based wor...
New
jimsynz
Beam Bots (or just BB for short) is a framework for building fault-tolerant robotics applications in Elixir using familiar OTP patterns. ...
New
Dmk
Xamal is a deployment tool for Elixir apps that deploys native releases to bare metal servers over SSH. It’s a port of GitHub - basecamp/...
New
Damirados
Hello everyone. After busy few months I am happy to announce v0.1.0 of Emerge & Solve. They are GUI (Emerge) and State management (S...
New
netoum
Corex is an accessible, unstyled UI component library for Phoenix that integrates Zag.js state machines using Vanilla JavaScript and Live...
New

We're in Beta

About us Mission Statement

Options

Thread Display Mode




Thread Preview

Skip Thread Previews