Celery correct chain of groups of tasks

user_6397377 0
Viewed 223

I have a list of lists of strings, which are filenames:

chunks_list = [["file_1", "file_2"], ["file_3", "file_4", "file_5"], ...]

I need to process this files in celery tasks, so I have task which does it:

@celery_app.task
def process_file_task(filename):
    # do some staff with file 
    # (e.g. produce data to Kafka)

An I need to run inner list of files parallel. And outer lists should be sequential.

The way of processing should be next:

worker1: file_1
worker2: file_2

The next group starts after the first is finished:

worker1: file_3
worker2: file_4
worker3: file_5

I tried to run my tasks in this way:

sequence_tasks = []
for chunks in chunks_list:
    sequence_tasks.append(
        group([process_file_task.si(filename) for filename in chunks])
    )

tasks_chain = chain(
    tasks_group for tasks_group in sequence_tasks
)()
tasks_chain.get()

The problem is that I could have thousands of files. And this way of running tasks consumes all my free memory, even if there are only hundred of them. Please tell me how to design my workflow in correct way to not to get out of memory.

celery python-3.x python
1 Answers

You didn't mention the setup details:

  • Where are your workers running? docker? (k8s/ecs/..) ec2?
  • What's the worker concurrency?
  • Is that a must that every file's task (within the same group) will run on a separate worker?

There's always an upper-case - limit of how many workers you have. Let's say you have 10 workers, each worker with concurrency 1 and 1,000 tasks in a group. They will run almost in parallel (because you don't have 1,000 workers, only 10) but that's fine. Celery will ensure that none of the tasks from the second group will start until it finishes all those first 1,000. If the time it takes to run each task is equal, every worker will handle 100 tasks and then will move on to the next chunk. If the time it takes is different, Celery will optimize by assigning the next tasks to the next free worker.

In other words, you know what the setup is (let's say EC2 instance with 16GB and 8 cores). You also know (I hope) what's the maximum memory size a single task can take. If an average task takes 1GB RAM, you can run at most 16 tasks with parallelism 1 or 6 tasks with parallelism 2 etc.)

user_10112530
Related