Celery를 사용한 실시간 트위터 스트리밍 처리

08 Nov 2023

Celery

이번 포스트에서는 Celery를 사용하여 실시간 트위터 스트리밍 처리를 하는 방법에 대해 알아보겠습니다. Celery는 파이썬 기반의 분산 작업 큐 시스템으로, 대규모 데이터 처리 작업을 더 효율적으로 처리할 수 있도록 도와줍니다.

Celery 설정하기

Celery를 사용하기 위해서는 먼저 Celery를 설치하고 프로젝트에 추가해야 합니다. 다음 명령을 사용하여 Celery를 설치합니다.

pip install celery

설치가 완료되면 Celery를 설정해야 합니다. 프로젝트의 루트 디렉토리에 celery.py 파일을 생성하고 다음과 같이 설정합니다.

from celery import Celery

# Celery 인스턴스 생성
app = Celery('myapp', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

# 작업 정의
@app.task
def process_tweet(tweet):
    # 트윗 처리 로직 작성
    ...

위의 예제에서는 Redis를 사용하여 Broker와 Backend를 설정하였습니다. 필요에 따라 다른 메시지 브로커와 백엔드를 사용할 수도 있습니다. process_tweet 함수는 Celery 작업으로 등록되어, 백엔드에 결과를 저장합니다.

실시간 트위터 스트리밍 처리하기

Celery를 사용하여 실시간 트위터 스트리밍을 처리하려면 트위터 API를 사용하는 클라이언트를 작성해야 합니다. 이 클라이언트는 트위터 스트림에서 새로운 트윗을 받아오고, process_tweet 작업을 Celery에 전달하여 처리합니다.

import tweepy
from celery import chord
from .celery import app, process_tweet

# 트위터 API 인증 정보 설정
consumer_key = 'YOUR_CONSUMER_KEY'
consumer_secret = 'YOUR_CONSUMER_SECRET'
access_token = 'YOUR_ACCESS_TOKEN'
access_token_secret = 'YOUR_ACCESS_TOKEN_SECRET'

# 트위터 인증
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

# 트위터 스트리밍 처리
class TweetListener(tweepy.StreamListener):
    def on_status(self, status):
        # 트윗 처리를 Celery 작업으로 전달
        chord(process_tweet.s(status.text) | process_tweet.signature()).apply_async()

    def on_error(self, status_code):
        if status_code == 420:
            return False

# Celery 워커 실행
if __name__ == '__main__':
    stream_listener = TweetListener()
    stream = tweepy.Stream(auth=auth, listener=stream_listener)
    stream.filter(track=['keyword1', 'keyword2'])

위의 예제에서는 tweepy 라이브러리를 사용하여 Twitter API에 연결하고, TweetListener 클래스를 통해 새로운 트윗을 수신합니다. on_status 메서드에서는 받아온 트윗을 Celery 작업으로 전달하고, on_error 메서드에서는 API 에러를 처리합니다.

이제 Celery 워커를 실행하고 트위터 스트림을 실시간으로 처리할 수 있습니다. 필요에 따라 TweetListener 클래스의 track 변수에 원하는 키워드를 추가하거나 수정하여 원하는 트윗을 처리할 수 있습니다.

이렇게 Celery를 사용하여 실시간 트위터 스트리밍을 처리하는 방법을 알아보았습니다. Celery를 사용하면 대용량의 데이터 처리를 더 효율적으로 처리할 수 있으며, 비동기 작업을 통해 실시간 데이터 처리에도 유용한 도구입니다.

#hashtags #Celery #트위터