An extension for celery to dispatch large amount of subtasks within a main task
Project description
celery-dispatcher
An extension for celery to dispatch large amount of subtasks within a main task, and process the results separately.
Installation
pip install celery-dispatcher
Usage
NOTICE:
celery-dispatcher
use tls to store running info, so the pool implementation using coroutines(like eventlet/gevent) can not be used
Firstly, yield subtask and its parameters in tuple from the main task by following order:
task
: signatureargs
: tupple/listkwargs
: dictoptions
: dict
Then register the result handler for each subtask using signal:
from celery import shared_task
from celery_dispatcher import dispatch
from celery_dispatcher.signals import subtask_success
@shared_task
def sqrt(i):
return i * i
@dispatch
@shared_task
def calc():
for i in range(10):
yield sqrt, (i,)
@subtask_success.connect(sender=calc)
def handle_result(root_id, task_id, retval, **kwargs):
print(retval)
Or register in the decorator directly:
from celery import shared_task
from celery_dispatcher import dispatch
@shared_task
def sqrt(i):
return i * i
def handle_result(root_id, task_id, retval, **kwargs):
print(retval)
@dispatch(receiver=handle_result)
@shared_task
def calc():
for i in range(10):
yield sqrt, (i,)
Options
The dispatch
accepts the following parameters:
options
: dictreceiver
: callablebackend
: strauto_ignore
: bool
General settings
dispatcher_result_backend
Default: No result backend enabled by default.
The backend used to store subtask info. Can be one of the following:
- redis: Use Redis to store the results. See Redis backend settings.
dispatcher_batch_size
Default: 1000
The batch size of subtask dispatching, or the result retrieving.
dispatcher_poll_size
Default: 1000
The queue size of subtasks in result retrieving. celery-dispatcher
put the unfinished subtasks into a queue, polling it continuously for completion of subtasks.
dispatcher_subtask_timeout
Default: 3600
The default timeout in seconds before celery-dispatcher
gives up retrieving the result of each subtask.
dispatcher_progress_update_frequency
Default: 1
The number of steps to record. celery-dispatcher
only actually store the updated progress in the background at most every N steps.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for celery_dispatcher-1.0.5-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 1af2a11019d5227a3dec640b69c46980283c5051c097df97b66e93e6945b9b49 |
|
MD5 | 245737e856907ad626bcc1378241a885 |
|
BLAKE2b-256 | d35caa7982a02e42ca8a1f72df545f0870344de4ecfcc4432af9ed496cf118b0 |