Celery는 Python 애플리케이션에서 백그라운드 작업을 효율적으로 처리할 수 있게 돕는 강력한 분산 태스크 큐 시스템입니다. 웹 스크래핑 파이프라인과 같이 대량의 데이터를 처리할 때, 여러 개의 Celery 태스크가 동시에 실행되어 데이터베이스에 페이로드를 저장하는 시나리오가 흔합니다. 하지만 이때 각 태스크가 부분적으로 중복되는 데이터를 생성하거나 처리하게 되면, 비효율적인 데이터베이스 쓰기 작업과 데이터 중복 문제에 직면할 수 있습니다.
이 글에서는 이러한 문제 상황을 분석하고, 여러 Celery 태스크의 결과물을 효율적으로 병합(Merge)하고 중복을 제거(Deduplicate)하며, 최종적으로 데이터베이스에 대량으로(Bulk Write) 저장하는 최적의 솔루션을 제시합니다. 구글 검색 상위 노출을 목표로, 개발자들이 쉽게 이해하고 적용할 수 있는 고품질 기술 문서를 제공합니다.
당신은 웹 스크래핑 파이프라인을 구축하고 있으며, 스크래핑된 데이터를 정제하여 데이터베이스에 저장하는 작업을 Celery 태스크로 구현했습니다. 각 Celery 태스크는 가공된 데이터(행의 리스트 형태)를 데이터베이스에 직접 저장합니다. 문제는 다음과 같습니다.
여러 Celery 태스크가 실행되면서 부분적으로 중복되는 데이터(행)를 생성할 수 있습니다.
각 태스크가 개별적으로 데이터베이스에 쓰기 작업을 수행하여, 전체적인 쓰기 효율성이 떨어집니다.
궁극적으로는 여러 태스크의 결과물을 하나로 병합하고, 이 과정에서 중복된 데이터를 제거한 후, 단일 트랜잭션으로 데이터베이스에 대량 쓰기(Bulk Write)를 하고 싶습니다.
현재의 설계 방식으로는 불필요한 데이터베이스 부하, 잠재적인 데이터 중복 발생, 그리고 전반적인 시스템 성능 저하를 야기할 수 있습니다.
2. 명확한 발생 원인
이러한 문제 상황은 주로 다음과 같은 이유로 발생합니다.
분산된 중복 처리 로직의 부재: 각 태스크가 독립적으로 데이터를 처리하고 저장하기 때문에, 다른 태스크가 이미 처리했거나 처리할 데이터를 인지하지 못합니다. 이로 인해 같은 데이터가 여러 번 생성되거나 저장될 가능성이 커집니다.
데이터베이스 I/O의 비효율성: 각 태스크가 데이터베이스에 개별적으로 연결하고, 트랜잭션을 시작하며, 적은 수의 행을 한 번에 삽입하는 것은 네트워크 오버헤드와 데이터베이스 서버의 부하를 증가시킵니다. 특히 대량의 데이터를 처리할 때는 상당한 성능 저하를 초래합니다.
태스크 간 오케스트레이션 부족: Celery는 개별 태스크 실행에 매우 강력하지만, 여러 태스크의 결과물을 한곳에 모아 후처리하는 복합적인 워크플로우를 명시적으로 설계하지 않으면, 통합 및 최적화 단계에서 어려움을 겪게 됩니다.
데이터 무결성 관리의 어려움: 중복 데이터 삽입을 방지하기 위한 데이터베이스의 Unique 제약 조건이 없다면, 애플리케이션 레벨에서 중복 제거 로직이 필요하며, 이를 분산된 여러 태스크가 동시에 안전하게 수행하기는 어렵습니다.
결론적으로, 핵심 원인은 분산된 환경에서 중복 데이터를 효율적으로 관리하고, 데이터베이스 쓰기 작업을 최적화하기 위한 통합적인 전략이 없기 때문입니다.
3. 해결 방법 및 코드 예시
Celery에서 여러 태스크의 결과를 병합하고 중복 제거, 벌크 쓰기를 수행하는 가장 효과적인 방법은 Celery Canvas의 강력한 기능을 활용하는 것입니다. 특히 group과 chord를 조합하여 여러 전처리 태스크의 결과를 단일 후처리 태스크로 모으는 전략이 유용합니다.
해결 방법 1: Celery Canvas (Group & Chord)를 이용한 작업 오케스트레이션
이 방법은 여러 스크래핑 또는 처리 태스크를 병렬로 실행하고, 모든 태스크가 완료되면 그 결과를 하나의 콜백 태스크로 전달하여 병합, 중복 제거, 그리고 벌크 쓰기를 수행하게 합니다.
from celery import Celery, group, chord
from celery.utils.log import get_task_logger
import time
# Celery 앱 설정 (실제 환경에 맞게 브로커와 백엔드 URL을 조정하세요)
app = Celery('my_scraper', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')
logger = get_task_logger(__name__)
@app.task
def scrape_data_part(part_id):
"""
데이터의 특정 부분을 스크래핑하거나 처리하는 예시 태스크입니다.
이 태스크는 중복이 포함될 수 있는 행 리스트를 반환합니다.
"""
logger.info(f"스크래핑 작업 {part_id} 시작...")
# 실제 환경에서는 웹에서 데이터를 가져오거나 복잡한 처리 로직을 수행합니다.
# 여기서는 예시를 위해 일부러 중복이 포함된 데이터를 생성합니다.
if part_id == 1:
return [{'id': 1, 'value': 'Apple'}, {'id': 2, 'value': 'Banana'}]
elif part_id == 2:
return [{'id': 2, 'value': 'Banana'}, {'id': 3, 'value': 'Cherry'}]
elif part_id == 3:
return [{'id': 3, 'value': 'Cherry'}, {'id': 4, 'value': 'Date'}]
return []
@app.task
def merge_and_save_data(results):
"""
여러 스크래핑 태스크의 결과를 병합하고 중복을 제거한 후,
데이터베이스에 대량으로 저장하는 콜백 태스크입니다.
"""
logger.info("모든 스크래핑 결과 수집 완료. 병합 및 저장 시작...")
all_rows = []
# 각 태스크에서 반환된 행 리스트들을 하나의 리스트로 통합
for sublist in results:
all_rows.extend(sublist)
# 중복 제거 로직 (예: 'id' 필드를 기준으로 고유한 데이터만 남김)
# 딕셔너리를 사용하여 O(N) 또는 O(N log N)의 효율성으로 중복 제거
deduplicated_rows_map = {}
for row in all_rows:
# 'id' 필드를 고유 키로 사용하여 마지막으로 발견된 행을 저장
deduplicated_rows_map[row['id']] = row
final_payload = list(deduplicated_rows_map.values())
logger.info(f"병합 및 중복 제거된 최종 데이터 개수: {len(final_payload)}")
# 여기서 데이터베이스 벌크 쓰기 로직을 구현합니다.
# 예: Django ORM의 YourModel.objects.bulk_create(final_payload, ignore_conflicts=True)
# SQLAlchemy의 session.bulk_insert_mappings(YourModel, final_payload)
# 실제 데이터베이스 저장을 시뮬레이션하기 위해 로깅합니다.
logger.info("--- 데이터베이스에 저장될 최종 페이로드 ---")
for row in final_payload:
logger.info(f"DB에 저장: {row}")
logger.info("----------------------------------")
return f"성공적으로 {len(final_payload)}개의 데이터를 병합 및 저장했습니다."
# 메인 실행 로직 예시 (Celery 워커가 실행된 상태에서 Python 스크립트로 실행)
# 이 부분은 블로그 게시물의 코드 블록 외부에서 설명되어야 합니다.
# if __name__ == '__main__':
# # 3개의 스크래핑 태스크를 병렬로 실행할 그룹 생성
# header = group(scrape_data_part.s(i) for i in range(1, 4))
#
# # 모든 그룹 태스크가 완료되면 실행될 콜백 태스크
# callback = merge_and_save_data.s()
#
# # group과 callback을 연결하여 chord 실행
# # chord(header)(callback)은 header의 모든 태스크가 완료된 후,
# # 그 결과들을 리스트로 묶어 callback 태스크에 인자로 전달합니다.
# result_async = chord(header)(callback)
# logger.info(f"작업 시작. 최종 결과 대기 중...")
#
# # 결과가 완료될 때까지 블로킹 (실제 프로덕션에서는 비동기적으로 처리하는 경우가 많습니다)
# print(result_async.get())
# logger.info("모든 작업 완료.")
코드 설명:
scrape_data_part 태스크는 실제 스크래핑 또는 데이터 처리 로직을 담당하며, 중복이 있을 수 있는 행 리스트를 반환합니다.
group(scrape_data_part.s(i) for i in range(1, 4))는 여러 scrape_data_part 태스크를 동시에 실행하도록 묶어줍니다.
merge_and_save_data는 chord의 콜백 태스크로, group에 속한 모든 scrape_data_part 태스크가 완료되면 그 결과물 리스트를 인자로 받아 실행됩니다.
merge_and_save_data 내에서는 모든 서브 태스크의 결과를 하나의 리스트로 통합한 후, 딕셔너리를 활용하여 효율적으로 중복을 제거합니다. (예: id 필드를 고유 키로 사용).
최종적으로 중복이 제거된 데이터 리스트를 가지고 데이터베이스의 벌크 쓰기 기능을 활용하여 한 번에 삽입합니다. 이는 개별적인 INSERT 문을 여러 번 실행하는 것보다 훨씬 효율적입니다.
해결 방법 2: 임시 저장소 활용 및 주기적/트리거 벌크 쓰기
이 방법은 각 Celery 태스크가 처리한 데이터를 즉시 DB에 쓰지 않고, Redis, Kafka, RabbitMQ 등과 같은 임시 저장소(큐 또는 캐시)에 쌓아두는 방식입니다. 그리고 특정 조건(예: 일정 시간 간격, 특정 데이터량 도달)이 되면 별도의 Celery 태스크(벌크 쓰기 태스크)를 트리거하여 임시 저장소의 데이터를 가져와 병합, 중복 제거 후 DB에 벌크 쓰기를 수행합니다.
# from celery import Celery
# from celery.utils.log import get_task_logger
# import redis # Redis 클라이언트 (예: `pip install redis`)
# import json
# app = Celery('my_scraper_queue', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')
# logger = get_task_logger(__name__)
# redis_client = redis.StrictRedis(host='localhost', port=6379, db=2) # 임시 저장소로 Redis 사용
@app.task
def process_and_add_to_queue(payload_data):
"""
데이터를 처리한 후 즉시 DB에 저장하지 않고, 임시 큐에 추가하는 태스크입니다.
"""
logger.info(f"데이터 처리 후 임시 큐에 추가: {payload_data}")
# Redis 리스트에 JSON 형태로 데이터를 저장
redis_client.rpush('temp_payload_queue', json.dumps(payload_data))
@app.task
def bulk_save_from_queue():
"""
임시 큐에 쌓인 데이터를 주기적으로 가져와 병합, 중복 제거 후 DB에 저장하는 태스크입니다.
이 태스크는 Celery Beat를 통해 일정 간격으로 스케줄링될 수 있습니다.
"""
logger.info("임시 큐에서 데이터 가져와 벌크 저장 작업 시작...")
all_payloads = []
# Redis 리스트에서 모든 데이터 가져오기 (원자적으로 처리하기 위해 LPUSH/LPOP 사용)
while True:
item = redis_client.lpop('temp_payload_queue')
if not item:
break
all_payloads.append(json.loads(item))
if not all_payloads:
logger.info("임시 큐에 저장할 데이터가 없습니다.")
return "저장할 데이터 없음"
# 중복 제거 로직 (해결 방법 1과 동일)
deduplicated_rows_map = {}
for row in all_payloads:
deduplicated_rows_map[row['id']] = row # 'id'는 고유 키 가정
final_payload = list(deduplicated_rows_map.values())
logger.info(f"큐에서 가져와 병합 및 중복 제거된 최종 데이터 개수: {len(final_payload)}")
# 여기서 데이터베이스 벌크 쓰기 로직 구현
logger.info("--- 큐 데이터를 DB에 벌크 저장 ---")
for row in final_payload:
logger.info(f"DB에 저장: {row}")
logger.info("----------------------------------")
return f"큐에서 성공적으로 {len(final_payload)}개의 데이터를 병합 및 저장했습니다."
# 이 bulk_save_from_queue 태스크는 Celery Beat 설정을 통해
# 예를 들어 매 1분마다 실행되도록 스케줄링할 수 있습니다.
# (예: celery_app.conf.beat_schedule = { 'bulk-save-every-minute': { 'task': 'your_module.bulk_save_from_queue', 'schedule': crontab(minute='*') } })
코드 설명:
process_and_add_to_queue 태스크는 처리된 데이터를 즉시 DB에 쓰지 않고, redis_client.rpush('temp_payload_queue', json.dumps(payload_data))를 통해 Redis 리스트 같은 임시 저장소에 쌓아둡니다.
bulk_save_from_queue 태스크는 Celery Beat와 같은 스케줄러에 의해 주기적으로 실행됩니다.
이 태스크는 임시 큐에서 모든 데이터를 가져와 (redis_client.lpop 사용), 해결 방법 1과 동일한 중복 제거 로직을 수행한 후 DB에 벌크 쓰기를 합니다.
이 방법은 데이터 처리와 DB 쓰기를 분리하여 시스템 부하를 분산하고, 트래픽이 불규칙하거나 지연 처리가 허용되는 환경에서 유연성을 제공합니다.
4. 향후 예방을 위한 팁
이러한 문제 발생을 사전에 방지하고 Celery 기반의 데이터 처리 파이프라인을 더욱 견고하게 구축하기 위한 몇 가지 팁입니다.
멱등성(Idempotency) 있는 태스크 설계: Celery 태스크는 네트워크 문제나 시스템 오류로 인해 여러 번 재시도될 수 있습니다. 동일한 태스크가 여러 번 실행되더라도 데이터베이스에 같은 데이터가 중복으로 삽입되거나 예기치 않은 부작용이 발생하지 않도록 멱등성을 고려하여 설계해야 합니다. 데이터베이스의 UNIQUE 제약 조건, 또는 삽입 전 데이터 존재 여부 확인 로직 등을 활용할 수 있습니다.
데이터베이스 스키마 최적화: 데이터베이스 테이블에 고유 식별자(Unique Key)를 적절히 설정하고, 필요한 경우 인덱스를 추가하여 중복 검사 및 데이터 삽입 성능을 향상시킵니다. bulk_create 시 ignore_conflicts=True 옵션 등을 활용하면 충돌 발생 시 오류 없이 넘어갈 수 있습니다.
모니터링 및 로깅 강화: Celery 작업의 성공/실패 여부, 처리된 데이터의 양, 중복 제거된 데이터의 비율, 그리고 데이터베이스 쓰기 시간 등을 상세하게 로깅하고 모니터링 시스템에 연동하세요. 이를 통해 잠재적인 성능 병목 현상이나 데이터 불일치 문제를 조기에 발견하고 대응할 수 있습니다.
작업 부하 분산 및 큐 관리: 너무 많은 데이터가 한 번에 하나의 태스크에 몰리거나, 큐에 미처리 작업이 과도하게 쌓이지 않도록 작업 부하를 적절히 분산하고 큐 길이를 모니터링하세요. Celery의 rate_limit 기능을 활용하거나, 서로 다른 목적의 큐를 사용하는 것도 좋은 방법입니다.
재시도(Retry) 전략 신중하게 수립: 데이터베이스 연결 오류나 일시적인 문제에 대비하여 Celery 태스크의 재시도(Retry) 전략을 신중하게 설정합니다. 하지만 무한정 재시도하거나 너무 짧은 간격으로 재시도하는 것은 시스템 부하를 가중시킬 수 있으므로, 적절한 백오프(backoff) 전략을 포함해야 합니다.
이러한 팁들을 통해 Celery 기반의 웹 스크래핑 및 데이터 처리 파이프라인을 더욱 견고하고 효율적으로 운영할 수 있을 것입니다. 올바른 설계와 최적화는 장기적으로 시스템의 안정성과 성능에 큰 영향을 미칩니다.
![[에러 해결] [Python asyncio] Redis PUB/SUB 장기 실행 TaskGroup 사용의 함정과 해결책](https://dev-error.com/wp-content/plugins/contextual-related-posts/default.png "[에러 해결] [Python asyncio] Redis PUB/SUB 장기 실행 TaskGroup 사용의 함정과 해결책")