More complex examples

[geopet85]

Hi, thank you for the efforts you put on the project. I personally use it for a toy ETL project I am currently working on, but I think more complex examples and topics are needed to convince people to start using it. 😄
Some suggestions:

• Could you add a more production-like setup and usage? Some complex functions besides printing text but doing some actual asynchronous function.
• Could you please add an example on the usage of RetryWithBackoffEntrypointExecutor? The documentation page seems incomplete without one.

[janbjorge]

Hi @geopet85 ,

Thanks for the kind words and for sharing your feedback! It’s awesome to hear you’re using PGQueuer for your ETL project.

• Production-like setup: I totally get the need for more complex, real-world examples. My use case is pretty simple, so I’ll try to come up with something meaningful without making it feel too “made up.” Maybe something like API calls or database interactions could work.
• RetryWithBackoffEntrypointExecutor example: You’re spot on—this needs more explanation. I’ll prioritize adding an example to show how it works in a retry scenario.

Appreciate the suggestions! If you’ve got specific ideas or use cases in mind, let me know, and I’ll see what I can do.

Cheers,
JB

[geopet85]

Appreciate the suggestions! If you’ve got specific ideas or use cases in mind, let me know, and I’ll see what I can do.

I think the biggest issue I had so far was handling of db restarts. The pgqueuer couldn't reconnect or tasks were stuck on picked state (because db restart was in the middle of a task run) and couldn't move forward or reset anyhow except with manual intervention. I saw that you try to address those in other PRs, feel free to include proper examples in the documentation or examples about it.
In general anything that makes pgqueuer more robust to hard or intermittent failures will be appreciated.

As for the CLI, I believe a find/filter function will strengthen its usability. e.g. I want to search all running tasks for a particular search text (some parameter in the payload). Similar to task filtering in flower.

[wlipski]

@geopet85 we are build some kind of a metrics collection service (primarily to collect metrics from different databases)

So far after that thing was fixed we moved forward and haven't experienced any issues.

Most probably that is because pgqueuer is running as a separate container (api to retrieve any metric related data and/or trigger a metric collection and worker to collect them)

If you'll be interested in the setup - let me know

And maybe, worth to mention:

We've observed that jobs could be in the picked state even after they died and pqgueuer reloaded itself. I think that is because of the heartbeat of the job. We've omit this by adding a separate schedule which just triggers "empty" job which will be picked by the pgqueur and it will move those "picked" jobs that are "dead" back to queued state because of the heartbeat.

By it will move I mean that pgqueuer checks the heartbeat and kind will reset jobs status (if heartbeat exceeds the threshold) it when it's picking next jobs.

@janbjorge maybe that is something that should be built in (maybe it's already exists and we just missed something)

[joej]

I'm using pgqueuer for a massive data ETL processing:

• django with management command to "enqueue" the data dump ("Collection" model), which does per-file job enqueuing based on a manifest file for each section of the dataset
• management command to "run_work" with multiple data dump processing entry points (e.g., process the "pubmed" data dump, etc) - dynamic entry point functions
• per-file tracking of success (Collection has FileProgress models), so I can monitor progress and know when the complete set (days, maybe months, of processing) are complete
• somewhat like pgskewer, a kind of "pipeline" - process pass1, then pass2, etc. simple ordering

Once I do have the multiple data sets (of similar data) loaded, a future processing logic will use those all to contribute (ETL) into a more agnostic data model for them all. That is: each data dump/database record has a provenance that has date, time, where collected, etc. and any of the presented data (aggregated from the separate elements) can trace back to: what data, from where, when loaded, when collected, from what specific location, and any external reference

complex and kicking my butt right now, but progressing.

I'd be happy to contribute, if there was something you can use from my experience/work with this.

[janbjorge]

Hey @joej,

Thanks for sharing this! Your use case sounds like a solid production setup.

A few things I'd love to learn more about:

1. Dynamic entry points - You mentioned "dynamic entry point functions" for processing different data dumps (like the pubmed example). Are you registering entry points at runtime, or do you have a set of predefined handlers that get selected based on the data dump type?

2. Pipeline ordering - You described the pass1 → pass2 sequential processing as "somewhat like pgskewer". How are you handling the dependencies between passes? Are you enqueuing pass2 jobs from within the pass1 completion handler, or using some other mechanism?

3. Scale - You mentioned tracking progress across "days, maybe months, of processing" with the FileProgress models. How many jobs are we talking about roughly? Any observations on performance or bottlenecks you've run into?

4. Provenance tracking - The future aggregation step where you trace data back through the entire pipeline (date, time, where collected, from what specific location, external references) sounds cool. Is this something you're building alongside PgQueuer in your Django models?

Would definitely be interested in any contributions or even just more details about your setup. If you're open to it, a write-up or example of how you've structured this could be valuable for others tackling similar ETL problems.

Cheers,
JB

[joej]

Sorry, but i had to move to something with features supporting more complexity ( SAQ).
That project is paused, while I take a break to build the monitoring infrastructure for it (also using SAQ, but with less complex requirements, so far).

1- each data (django) app would dynamically register its entry point

2- this had been the problematic area: sequential "pass" functions. I had to keep building out infrastrucure in the django project: a "manifest" model where passes would update when done, and other fields capturing state logic. So, this drove me towards the more complex task running queue, but trying to stay away from celery, rabbitmq, redis). I ended up with valkey/redis and SAQ anyway.

I had some logic near the task worker code that had the list of functions for "pass1" and "pass2" etc, so that there could be some validation or orchestration of sequencing. It just wasn't enough.

3- One data dump would have different data schema types (people, papers, organizations) and subdirectories of data (jsonl.gz) files. I ended up tasking, processing, and tracking each data file. So, 5000+ files in 1 data dump and 3000+ in the next data dump, etc.

Any failure to process then SQL update... on conflict ... if ... would require me reprocessing the whole file. There would be nearly 240 million data records in total with billions of relational records between them

4- The provenance is for citation, so the data dump "collection" (django model) would have a date, source, explanation and the file manifest.

The processing and loading would ensure each record had the Collection id stored. The task processing functions would "check off" each file as it completed with additional info as a FilesProgress type django object/record (i.e., records data, not relational data). The pass2 could get triggered from the completion of files enough to trigger the "do the relational connections, now that the records all exist to connect".

I did start to add logic to track minimum required record types: e.g., All of the "documents" and "authors" need to be done before the relational Many2Many model DocumentsAuthors relations could be created.

Again, this is where the "graph" of what connects to what was causing me to write more code outside of pgqueuer than I would have to in another task runner.