[python] 다중 처리를 사용하는 셀러리 병렬 분산 작업

당신의 목표는 :

1. 작업을 여러 머신에 배포 (분산 컴퓨팅 / 분산 병렬 처리)
2. 모든 CPU (다중 처리 / 스레딩)에 지정된 컴퓨터의 작업을 배포합니다.

셀러리는이 두 가지를 매우 쉽게 할 수 있습니다. 가장 먼저 이해해야 할 것은 각 셀러리 작업자가 기본적 으로 시스템에서 사용 가능한 CPU 코어 수만큼 작업을 실행 하도록 구성 되어 있다는 것입니다.

동시성은 작업을 동시에 처리하는 데 사용되는 프리 포크 작업자 프로세스의 수입니다. 이러한 모든 작업이 바쁘면 새 작업이 처리되기 전에 작업 중 하나가 완료 될 때까지 기다려야합니다.

기본 동시성 수는 해당 시스템 (코어 포함)의 CPU 수이며 -c 옵션을 사용하여 사용자 지정 번호를 지정할 수 있습니다. 최적의 수는 여러 요인에 따라 달라 지므로 권장되는 값은 없지만 작업이 대부분 I / O 바운드 인 경우이를 늘릴 수 있습니다. 실험에 따르면 CPU 수를 두 배 이상 추가하는 것은 거의 발생하지 않습니다. 효과적이고 대신 성능을 저하시킬 가능성이 있습니다.

즉, 각 개별 작업은 다중 CPU / 코어를 사용하기 위해 다중 처리 / 스레딩을 사용하는 것에 대해 걱정할 필요가 없습니다. 대신 셀러리는 사용 가능한 각 CPU를 사용하기에 충분한 작업을 동시에 실행합니다.

그 과정에서 다음 단계는 .NET Framework의 일부 하위 집합 처리를 처리하는 작업을 만드는 것 list_of_millions_of_ids입니다. 여기에는 몇 가지 옵션이 있습니다. 하나는 각 작업이 단일 ID를 처리하도록하는 것이므로 N 개의 작업을 실행합니다 N == len(list_of_millions_of_ids). 이렇게하면 한 명의 작업자가 일찍 끝나고 그냥 기다리는 경우가 없기 때문에 작업이 모든 작업에 균등하게 분배됩니다. 작업이 필요한 경우 대기열에서 ID를 가져올 수 있습니다. 셀러리를 사용하여 (John Doe가 언급했듯이) 이것을 할 수 있습니다 group.

tasks.py :

@app.task
def process_id(item):
    id = item #long complicated equation here
    database.objects(newid=id).save()

그리고 작업을 실행하려면 :

from celery import group
from tasks import process_id

jobs = group(process_id.s(item) for item in list_of_millions_of_ids)
result = jobs.apply_async()

또 다른 옵션은 목록을 작은 조각으로 나누고 그 조각을 작업자에게 배포하는 것입니다. 이 접근 방식은 일부 작업자가 작업을 계속하는 동안 대기중인 작업자가있을 수 있기 때문에 일부주기를 낭비 할 위험이 있습니다. 그러나 셀러리 문서 에서는 이러한 우려가 종종 근거가 없다고 말합니다 .

일부 사람들은 작업을 청킹하면 병렬 처리가 저하 될 것이라고 걱정할 수 있지만 바쁜 클러스터에서는 거의 해당되지 않으며 메시징 오버 헤드를 피하고 있기 때문에 성능이 크게 향상 될 수 있습니다.

따라서 목록을 청크하고 각 작업에 청크를 배포하면 메시징 오버 헤드가 줄어들어 성능이 더 우수하다는 것을 알 수 있습니다. 한 번에 하나의 ID를 수행하는 대신 각 ID를 계산하고 목록에 저장 한 다음 DB에 전체 목록을 추가하는 방식으로 데이터베이스에 대한 부하를 약간 줄일 수도 있습니다. . 청크 접근 방식은 다음과 같습니다.

tasks.py :

@app.task
def process_ids(items):
    for item in items:
        id = item #long complicated equation here
        database.objects(newid=id).save() # Still adding one id at a time, but you don't have to.

작업을 시작하려면 :

from tasks import process_ids

jobs = process_ids.chunks(list_of_millions_of_ids, 30) # break the list into 30 chunks. Experiment with what number works best here.
jobs.apply_async()

어떤 청킹 크기가 최상의 결과를 제공하는지 약간 실험 할 수 있습니다. 메시지 오버 헤드를 줄이면서 작업자가 다른 작업자보다 훨씬 빨리 청크를 완료하고 할 일이없는 상태로 대기하는 일이 없도록 크기를 충분히 작게 유지하는 최적의 지점을 찾고 싶습니다.