rq
rq copied to clipboard
rq
Group jobs into batches and retrieve by batch name
A common pattern for us when using RQ is to iterate over an array of function arguments to pass to the queue.enqueue method. One issue we're having is then being able to query RQ to view the status for that "batch" of jobs represented by the arg iterable. We can pass the array of job IDs into the fetch_many method but people have been saying it's too cumbersome to track this large array of IDs across different processes and infrastructure. I was wondering whether it would be feasible to add an additional field to the Job object in Redis that could be called "batch_id", which would be set by the user and could be passed into the fetch_many method as an alternative to an array of job IDs. This would also make it easier to query all of the logs from that batch from our logging backend. The pattern may look like so:
batch_id = str(uuid.uuid4())
inputs = ["foo", "bar"]
for i in inputs:
queue.enqueue(func, i, batch_id=batch_id)
jobs = queue.fetch_many(batch_id=batch_id, connection=redis) # Returns array of Job instances
Would a PR be welcome for this? And if so is there any specific part of the codebase I should start with?
I like this idea and it seems simple enough to do with storing a redis set of job ids.
Although I would suggest making batches be a class just like Job.
batch = Batch([job1, job2])
batch.add(job3)
batch_id = batch.get_id()
...
batch = Batch.fetch(batch_id)
jobs = batch.get_jobs()
That way having a Batch object could open the door to having more functionality such as getting the overall status of the batch. ("failed" == some job failed, "finished" == all jobs finished, "started" == all jobs started/queued/deferred)
On the surface this looks like a good idea, but I think the implementation may be a bit tedious. We'll need to keep Batch objects on Redis that stores a list of job IDs. Coordinating the deletion of Batch objects when jobs get deleted or expired is also something we need to think about.
I agree it would be tedious and have tricky things to deal with. (I'm currently trying to write something similar to this for managing my custom jobs and keep running into walls.) But it would be great since this would be like celery groups or the currently implemented pipelines, but much better since you can continuously add new jobs instead of having to have everything set up ahead of time.
I would imagine you could have a .close() function which will tell rq that we no longer care for getting the results of the jobs and we can remove the job list. With a fallback of automatically closing after some time has passed. Similar to job results ttl.
When the worker finishes a job, could it pull the batch ID and extend the TTL of the batch by some user-specified value? This means as long as the batch’s jobs continue to be processed, the batch will persist, but once it finishes all of them, the batch expires. Additionally, when we start a new job we could remove the expiry from the batch to make sure it doesn’t expire while the job is running.
When the worker finishes a job, could it pull the batch ID and extend the TTL of the batch by some user-specified value? This means as long as the batch’s jobs continue to be processed, the batch will persist, but once it finishes all of them, the batch expires. Additionally, when we start a new job we could remove the expiry from the batch to make sure it doesn’t expire while the job is running.
Yes, so essentially this is what RQ does when maintaining job keys with TTLs, so we'll need to do this for each batch, on a new BatchRegistry. I'm ok if someone wants to take ownership of this, but this feature will require a lot of work :).
Sounds great, I’m happy to take a stab at it.
Please don't hesitate to ask for review in small batches, to make sure that the general direction is correct so you don't end up rewriting large chunks of code.