Celery를 이용한 크롤링 예제

개요

Airflow와 Celery를 비교했다.

크롤링 관리 Airflow vs Celery

 

크롤링 진행 시, Celery를 사용하는 게 더 적합하다는 판단과 함께 예제를 작성했다.

 

FastAPI + Celery + requests + BeautifulSoup 를 활용한 크롤링 작업을 수행하는 기본적인 방법을 예제

크롤링은 백그라운드에서 실행되도록 진행

기본 구조

• FastAPI: 사용자가 API를 통해 크롤링 요청을 보냄
• Celery: 요청을 큐에 넣고, 백그라운드에서 크롤링 실행
• Redis: Celery의 메시지 브로커 역할 (Task 관리)
• requests + BeautifulSoup: 웹사이트에서 HTML 데이터를 가져와서 파싱

FastAPI

FastAPI에서 사용자가 크롤링을 요청하면 Celery 태스크를 실행하는 API

from fastapi import FastAPI, BackgroundTasks
from celery_tasks import crawl_website_task, celery_app
from celery.result import AsyncResult

app = FastAPI()

@app.get("/")
def home():
    return {"message": "Welcome to the Celery Web Scraper API"}

@app.post("/crawl")
def start_crawling(url: str, background_tasks: BackgroundTasks):
    task = crawl_website_task.delay(url)  # Celery 비동기 태스크 실행
    return {"task_id": task.id, "message": f"Crawling started for {url}"}

@app.get("/task/{task_id}")
def get_task_status(task_id: str):
    
    if celery_app.conf.result_backend is None or celery_app.backend is None:
        return {"error": "Celery backend is disabled in FastAPI"}

    task_result = AsyncResult(task_id, app=celery_app)

    return {
        "task_id": task_id,
        "status": task_result.status,
        "result": task_result.result
    }

• start_crawling(url): 사용자가 특정 URL 크롤링을 요청하면 Celery 태스크 실행 (delay(url))
• get_task_status(task_id): Celery 태스크의 현재 상태를 조회

Celery 크롤링 태스크

Celery에서 크롤링 작업을 실행하는 코드

import requests
from bs4 import BeautifulSoup
from celery import Celery

# Celery 인스턴스 생성
celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",  # Redis 메시지 브로커 사용
    backend="redis://localhost:6379/0"  # 작업 결과 저장
)

# Celery 설정 추가
celery_app.conf.update(
    result_backend="redis://localhost:6379/0",
    result_expires=3600,  # 결과 1시간 유지
    task_ignore_result=False,  # 결과 저장 활성화
    broker_connection_retry_on_startup=True
)

@celery_app.task(bind=True)
def crawl_website_task(self, url: str):
    """ 주어진 URL을 크롤링하여 제목(title)과 링크(a 태그)를 추출하는 작업 """
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()  # HTTP 오류 발생 시 예외 처리

        soup = BeautifulSoup(response.text, "html.parser")
        title = soup.title.string if soup.title else "No title found"

        # 모든 링크(a 태그) 가져오기
        links = [a["href"] for a in soup.find_all("a", href=True)]

        return {"title": title, "links": links[:5]}  # 상위 5개 링크 반환

    except requests.RequestException as e:
        return {"error": str(e)}

• celery_app.task: Celery 태스크를 정의
• crawl_website_task(url): 주어진 URL을 크롤링하고 제목과 링크를 추출

# app 실행
uvicorn app:app --host 0.0.0.0 --port 8000

Redis와 Celery 실행

(1) Redis 실행 (메시지 브로커)

redis-server

 

(2) Celery 실행 (Worker 실행)

celery -A celery_tasks.celery_app worker --loglevel=info

이렇게 하면 Celery worker가 대기 상태에서 태스크를 실행할 준비를 한다.

API 요청 및 실행 테스트

FastAPI docs를 통해서 테스트 진행 혹은 curl 사용

 

(1) 크롤링 요청 보내기

curl -X POST "http://127.0.0.1:8000/crawl?url=https://example.com"

 

Response:

{
    "task_id": "c4256b26-b36e-48cf-bc5e-b6a77a3f263b",
    "message": "Crawling started for https://example.com"
}

 

(2) 크롤링 결과 확인

curl -X GET "http://127.0.0.1:8000/task/c4256b26-b36e-48cf-bc5e-b6a77a3f263b"

Response 성공:

{
    "task_id": "c4256b26-b36e-48cf-bc5e-b6a77a3f263b",
    "status": "SUCCESS",
    "result": {
        "title": "Example Domain",
        "links": ["https://www.iana.org/domains/example"]
    }
}

 

Response 실행 중:

{
    "task_id": "c4256b26-b36e-48cf-bc5e-b6a77a3f263b",
    "status": "PENDING",
    "result": null
}

Celery를 활용한 추가 기능

Celery를 활용하면 크롤링 서비스를 더 확장할 수 있다.

✅ 스케줄링

일정 시간마다 크롤링 실행하고 싶은 경우 celery-beat 사용 가능

from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "crawl_every_minute": {
        "task": "celery_tasks.crawl_website_task",
        "schedule": crontab(minute="*/1"),  # 1분마다 실행
        "args": ("https://example.com",),
    },
}

 

✅ 동시 크롤링

Celery의 worker를 여러 개 실행하면 동시 크롤링 가능

celery -A celery_tasks.celery_app worker --loglevel=info --concurrency=4

→ --concurrency=4: 동시 4개의 크롤링 태스크 실행 가능

 

✅ 크롤링 실패 시 자동 재시도

@celery_app.task(bind=True, autoretry_for=(requests.RequestException,), retry_kwargs={"max_retries": 3})
def crawl_website_task(self, url: str):
    ...

 

→ retry_kwargs={"max_retries": 3}: 요청 실패 시 최대 3번 자동 재시도

---

✨ FastAPI 에서 background_tasks: BackgroundTasks

 

BackgroundTasks를 사용하면 요청을 보낸 후 백그라운드에서 실행되지만, FastAPI 자체의 스레드 내에서 실행되는 방식이기 때문에 제한이 있다. (완전히 독립적인 작업은 아님)

몇가지 단점들:

1. FastAPI 서버가 재시작되면 백그라운드 작업도 사라짐
   1. Celery 같은 외부 태스크 큐를 사용해야 안정적으로 실행 가능
2. 서버가 강제 종료되면 작업이 중단됨
   1. 예: uvicorn을 중지하면 진행 중이던 백그라운드 태스크도 종료됨
3. 스케일링(여러 개의 서버 인스턴스 실행) 시, 작업이 특정 인스턴스에만 남아있을 수 있음
   1. 로드밸런서를 사용할 경우 문제가 될 수 있음

즉, BackgroundTasks는 가벼운 작업(예: 로깅, 이메일 전송)에는 적합하지만, 장시간 실행되는 크롤링 작업에는 부적합하다. 장시간 실행되는 백그라운드 작업은 Celery를 사용하는 것이 안정적이다.