Celery Canvas 워크플로우: Chain, Group, Chord

여러 태스크를 순차·병렬·집계로 엮을 때, 콜백을 직접 잇지 않고 흐름을 선언으로 표현하는 방법을 다룹니다.

태스크를 직접 잇는 코드의 한계

비동기 태스크 큐에서 태스크 하나를 던지는 일은 간단합니다. 문제는 여러 태스크를 엮을 때 생깁니다. A가 끝나면 B, B 결과로 C를 병렬 실행하고, 모이면 D를 부르는 식입니다. 이 흐름을 콜백으로 직접 이으면 결과를 주고받는 코드가 워커마다 흩어집니다.

Canvas(캔버스)는 Celery의 워크플로우 조합 시스템입니다. 태스크를 Chain(순차), Group(병렬), Chord(분산·수집)로 묶어 멀티스텝 작업을 선언으로 표현합니다. 흩어진 콜백 대신, 흐름 자체를 하나의 값으로 들고 다니는 방식입니다.

조합의 최소 단위 Signature

Signature(시그니처)는 태스크 호출을 직렬화 가능한 딕셔너리로 포장한 객체입니다. 함수를 바로 실행하지 않고 "이 인자로 이 태스크를 부른다"는 계획만 들고 다닙니다. 모든 Canvas 조합은 이 Signature를 단위로 엮습니다.

python

from celery import signature
 
# 명시적 선언
sig = signature('myapp.tasks.add', args=(2, 2), options={'queue': 'default'})
 
# 축약 문법 (가장 일반적)
sig = add.s(2, 2)

.s()는 partial(부분 적용) 시그니처입니다. 비어 있는 첫 인자 자리에 앞 태스크의 결과가 채워집니다. 반대로 .si()는 immutable(불변) 시그니처로, 앞 결과를 무시하고 고정 인자만 씁니다.

python

# partial: 이전 결과를 첫 번째 인자로 받음
sig = mul.s(8)        # mul(previous_result, 8)
 
# immutable: 이전 결과 무시, 항상 고정값
sig = add.si(2, 2)    # 항상 add(2, 2) = 4

이 둘의 구분이 체인에서 결과가 전파될지 끊길지를 결정합니다. 결과를 이어 받아야 하면 .s()를 쓰고, 알림처럼 앞 값과 무관한 단계는 .si()를 씁니다.

Chain의 순차 실행

Chain(체인)은 태스크를 순서대로 실행하고, 각 결과를 다음 태스크의 첫 인자로 넘깁니다. 파이프 연산자 |로 잇거나 chain()으로 묶습니다.

python

from celery import chain
 
# 파이프 연산자 (권장)
workflow = add.s(4, 4) | mul.s(8) | mul.s(10)
 
# 명시적 chain()
workflow = chain(add.s(4, 4), mul.s(8), mul.s(10))
 
result = workflow()        # workflow.apply_async()와 동일
print(result.get())        # 640

계산은 ((4 + 4) × 8) × 10 = 640으로 흐릅니다. 중간 결과는 parent로 거슬러 올라가 조회합니다.

python

print(result.get())                 # 640  (최종)
print(result.parent.get())          # 64   (mul(8, 8))
print(result.parent.parent.get())   # 8    (add(4, 4))

한 단계가 실패하면 체인은 그 지점에서 멈추고 뒤 단계는 실행되지 않습니다. 그래서 체인은 앞 결과가 뒤 단계의 전제가 되는 흐름에 맞습니다.

Group의 병렬 실행

Group(그룹)은 여러 태스크를 동시에 발행하고 결과를 리스트로 모읍니다. 태스크 사이에 의존이 없을 때 씁니다.

python

from celery import group
 
# 10개 add 태스크 병렬 실행
tasks = group(add.s(i, i) for i in range(10))
result = tasks()
 
print(result.get())   # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

결과 순서는 발행 순서를 따르므로 인덱스로 항목을 되짚을 수 있습니다. 개별 태스크의 상태는 result.results로 하나씩 확인합니다.

python

for res in result.results:
    print(res.state, res.result)

그룹은 결과를 모으기만 할 뿐, 모은 뒤 한 번에 처리하는 단계는 포함하지 않습니다. 그 "모은 다음"을 흐름에 붙이는 것이 다음 Chord입니다.

Chord의 분산과 수집

Chord는 그룹을 병렬로 실행한 뒤, 모든 결과가 도착하면 콜백 하나를 실행합니다. 작업을 잘게 나눠 뿌리고(분산), 결과를 한곳에 모으는(수집) Map-Reduce 패턴입니다.

위 그림은 10개 add 태스크가 병렬로 실행된 뒤, 합산 콜백 하나로 결과가 모이는 흐름을 나타냅니다. Chord는 헤더(그룹)와 바디(콜백)의 조합이며, 아래 두 표현은 같은 동작을 합니다.

python

from celery import chord, group
 
@app.task
def xsum(values):
    return sum(values)
 
header = group(add.s(i, i) for i in range(10))
 
# 1) 명시적 chord
workflow = chord(header)(xsum.s())
 
# 2) 파이프 (group | callback)
workflow = (group(add.s(i, i) for i in range(10)) | xsum.s())()
 
print(workflow.get())   # 90

합계 90은 [0, 2, ..., 18]의 합과 같습니다. 그룹과 달리 Chord는 결과를 모으는 콜백까지 한 흐름으로 묶는다는 점이 다릅니다.

Chord의 동기화와 제약

Chord는 "모든 헤더가 끝났는가"를 판단해야 콜백을 발화합니다. 이 판단을 어떻게 하는지는 결과 백엔드(result backend, 태스크 결과를 저장하는 저장소)에 따라 갈립니다.

결과 백엔드	완료 판단 방식	비용
Redis	결과를 리스트에 RPUSH, 길이를 LLEN으로 확인	원자적 카운터, 폴링 불필요
데이터베이스	celery.chord_unlock이 매초 폴링	DB 쿼리 반복, 상대적으로 느림

Redis는 헤더가 하나 끝날 때마다 결과를 리스트에 넣고 길이를 세어, 예상 개수에 도달하면 즉시 콜백을 띄웁니다. 데이터베이스 백엔드는 celery.chord_unlock 태스크가 매초 결과 집합이 다 찼는지 확인하므로 지연과 부하가 큽니다. 이 동작 차이는 Celery가 결과 백엔드별로 다르게 구현한 Chord 동기화 방식에서 비롯합니다.

그래서 Chord에는 전제 조건이 있습니다. 아래 셋 중 하나라도 어기면 동작하지 않습니다.

result_backend가 반드시 설정돼 있어야 합니다.
헤더 태스크는 ignore_result=False여야 결과가 모입니다.
RPC 백엔드(rpc://)는 Chord를 지원하지 않습니다.

python

# RPC 백엔드에서는 chord가 실패합니다
app.conf.result_backend = 'rpc://'
workflow = (group(tasks) | callback)()   # 실패
 
# Redis 등 실제 백엔드에서 동작합니다
app.conf.result_backend = 'redis://redis:6379/1'

설정만 점검하면 흔한 Chord 오류는 사전에 걸러집니다. 헤더에 ignore_result=True가 섞이면 수집이 끝나지 않으니 주의합니다.

자주 쓰는 조합 패턴

세 프리미티브는 중첩해 더 큰 흐름을 만듭니다. 흔한 형태는 그룹을 체인 안에 넣는 ETL(추출·변환·적재, Extract-Transform-Load) 패턴입니다.

python

from celery import chain, group
 
workflow = chain(
    group(fetch.s(url) for url in urls),   # 병렬 추출
    transform.s(),                          # 변환
    load.s(),                               # 적재
)
print(workflow().get())   # {'count': 3}

반대로 체인 여러 개를 그룹으로 감싸면 독립 파이프라인을 병렬로 돌립니다. 각 URL이 fetch → parse → validate → store를 순차로 거치되, URL끼리는 동시에 처리됩니다.

python

workflows = group(
    chain(fetch.s(url), parse.s(), validate.s(), store.s())
    for url in urls
)

분기가 필요하면 self.replace()로 런타임에 워크플로우를 바꿉니다. Celery에는 if/else 프리미티브가 없어, 태스크 안에서 조건에 맞는 체인을 만들고 raise self.replace(...)로 자신을 교체합니다.

python

@app.task(bind=True)
def route_by_type(self, data):
    if data['type'] == 'urgent':
        workflow = chain(process_urgent.s(data), notify_admin.s())
    else:
        workflow = chain(process_normal.s(data), queue_for_review.s())
    raise self.replace(workflow)   # 반드시 raise로 호출

self.replace()는 반드시 raise와 함께 써야 현재 태스크가 새 워크플로우로 대체됩니다. 정적 조합만으로 표현하기 어려운 조건 분기를 이 방식으로 처리합니다.

에러 처리와 재시도

조합이 길수록 한 단계의 실패를 어떻게 다룰지가 중요합니다. 성공·실패 콜백은 link와 link_error로 붙입니다.

python

@app.task
def on_error(request, exc, traceback):
    """실패한 태스크의 요청·예외·트레이스백을 받습니다."""
    print(f"실패: {exc}")
 
# 체인 전체에 에러백 연결
workflow = (task1.s() | task2.s() | task3.s())
result = workflow.apply_async(link_error=on_error.s())

일시적 오류는 콜백보다 재시도가 먼저입니다. 네트워크 호출처럼 다시 시도하면 풀릴 작업에는 자동 재시도를 겁니다.

python

@app.task(
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,        # 1, 2, 4, 8초 대기
    retry_backoff_max=600,     # 최대 10분
    retry_jitter=True,         # 대기 시간 무작위화
    max_retries=5,
)
def call_api(url):
    return requests.get(url, timeout=10).json()

retry_backoff는 재시도 간격을 지수로 늘리고, retry_jitter는 그 간격을 무작위로 흩어 여러 태스크가 동시에 몰리는 것을 막습니다. 워크플로우가 커지면 거대한 단일 조합으로 묶지 않습니다. 단계별로 쪼개 발행하고, 객체 대신 ID를 넘겨 직렬화 위험을 줄이는 편이 복구에 유리합니다.

정리

Chain은 결과를 이어 순차로, Group은 의존 없는 작업을 병렬로, Chord는 병렬 결과를 콜백으로 모읍니다. 조합의 단위는 Signature이며, .s()는 앞 결과를 받고 .si()는 무시한다는 구분이 흐름의 전파를 결정합니다. Chord는 결과 백엔드가 완료를 판단하므로 result_backend 설정과 헤더의 ignore_result=False가 전제 조건입니다. 큰 워크플로우는 한 번에 엮기보다 단계로 분해하고 ID를 넘기는 편이 실패 복구에 유리합니다.