[Python][Celery] 병렬작업을 진행하는 방법

Celery의 병렬 작업을 쓰게 된 계기는 데이터 관련 처리를 하면서 전체 유저에 관한 엑셀 파일을 만드는 작업에서 시작되었다.

 

하나의 shell에서 작업하기에는 시간이 많이 걸려서 원래 쉘을 한 9개 정도 켜서 작업을 진행했었는데 이러한 작업들을 충분히 작업큐로 해결할 수 있을 것 같은 생각이 들었다. ( 정말 쉘 9개 켜서 복사 붙여넣기 하는 거 시간이 정말 아까웠다. 이걸로 인해 나의 개발 시간 1시간을 벌 수 있게 되었다.

 

그래서 병렬작업으로 처리하면서 어떻게 돌릴 수 있을까? 에서 시작되어 까먹지 않기 위해 이 글을 적으려고 한다.

 

 

 

celery가 돌아가는 기본 원리는 다들 아실거라 생각하고 진행하겠습니다.

 

@shared_task
def something_task(self):
	...
    


app.conf.beat_schedule = {
	'daily_07_am': {
    	'task': 'path.to.tasks.method',
        'schedule': crontab(minute=0,hour=22)
    },
    ...
}

 

기본적으로 다음과 같이 진행됩니다.

 

하지만 위에서 말한 사례처럼 쉘 9개 정도 켜야(병렬적으로 처리해야) 효율적으로 작업을 할 수 있는 상황입니다.

이러한 상황에서 어떤 방법을 사용하면 될까? 

 

https://docs.celeryproject.org/en/stable/userguide/canvas.html#groups

Canvas: Designing Work-flows — Celery 5.2.1 documentation

바로 group이라는 것을 사용하면 된다.

 

group을 사용하면 병렬적으로 처리를 할 수 있는데 이 상황은 쉘 9개 킨 상황과 매우 유사하다.

 

 

저는 파일을 만들고 이 파일에 대한 데이터를 병합하는 작업을 하고 있었기 때문에 작업을 for문을 돌려서 적절히 처리했다.

 

from celery import group

result = group(
			정기적으로_매일_추출하는_작업.s(
            	오늘날짜, 유저정보아이디들[CHUNK*i:CHUNK*(i+1)],i) for i in range(크기)
             )

RESULT = result()
assert RESULT.ready()

...

 

이렇게 해당 함수를 불러 일으켜서 처리를 하면 된다.