It is expected for you to write your own GenStage producers and plug them into Broadway. There is even a guide that covers this: Custom Producers — Broadway v1.0.7

However, the producers_consumers and consumers are Broadway responsibility, you can’t plug them in.

Create a buffer process. Generate a job for each sensor day combo you need. Queue up those jobs in your buffer. Create a Broadway producer for your buffer and in the Broadway pipeline pull the data, batch it and insert. Scale up producer consumers as needed.

My take is that there are some common producers provided, but the rest is up to us. Nothing wrong with building your own producers/consumers. Last time I read the docs, the takeway was to limit the number of producer-consumers like in GenStage.

(I’m assuming that you’re using Ecto). No, it has a pool of connections so you can query the DB in parallel. But then you need to check how much the DB is able to sustain.

You have two distinct actions, one is to schedule, the other is to fetch…

Schedule is something different, but You could have GenServer (maybe one per sensor), triggering event periodically, that will put job in a GenStage pipeline.

Then, the pipeline will achieve its work, even under load pressure. You will achieve concurrency by specifying the number of producer/consumer.

Beware of not going too fast, and not flooding your database.It happened to me when trying to crawl a website and saving to db. I was doing insert_all with potentially too much data, and postgrex did not like it that much.

GenStage helped me to have a better control of concurrency.

Flow would be nice if You have aggregation per sensor, but it seems not, as You just insert this into db.

Thanks José. I’m confused by this last statement though. Wouldn’t the consumer in my case be the layer to performs the bulk insert with ecto? That would be a consumer I would have to write.

My steps are:
(1) query → (2) for each, fetch (API get request) many times → (3) bulk insert

So I see this as producer → producer_consumer → consumer. Is this correct?