지난 시간 리뷰
저번 시간에는 Sequencial executor, local executor에 대해서 배웠었다.
결국에는 조금씩의 확장이 이루어지면서 executor는 발전한다.
local executor에서 발전된 executor는 무엇일까? 로컬에서 하나의 컴퓨터만 동작시키는 것이 아니라 worker 즉 컴퓨터가 더 많이 추가된 케이스가 될 것이다.
오늘은 그런 확장된 개념인 celery executor에 대해 알아보고자 한다.
Celery Executor!
우선 celery에 대해 간략히 설명하자면 task queue(message broker)에 작업을 순차적으로 적재하고 task들을 queue에서 하나씩 빼서 worker node(subscriber)에 할당하는 방식이다.
celery에 더욱 자세히 알고 싶다면 아래의 링크들을 참조하자.
https://docs.celeryq.dev/en/latest/getting-started/introduction.html
https://airflow.apache.org/docs/apache-airflow/stable/executor/celery.html
전반적인 아키텍쳐
실행 순서도
celery executor 실행환경 만들기
일단 실행환경을 만들기 위해서는 message queue에 사용할 redis에 대해 잠깐만 알아보자!
redis란?
Key, Value 구조의 비정형 데이터를 저장하고 관리하기 위한 오픈 소스 기반의 비관계형 데이터 베이스 관리 시스템 (DBMS)이다.
데이터베이스, 캐시, 메세지 브로커로 사용되며 인메모리 데이터 구조를 가진 저장소이다.
레디스는 아래와 같은 특징을 지니고 있다.
한 번에 하나의 명령만 처리할 수 있으며, Key-Value 구조이기 때문에 쿼리를 사용할 필요가 없다.
데이터를 디스크에 쓰는 구조가 아니라 메모리에서 데이터를 처리하기 때문에 속도가 빠른편이다.
우리는 message broker를 redis로 사용할 예정이다.
airflow_celery 설치하기
pip install 'apache-airflow[celery]'을 입력해 해당 프레임워크를 설치하자.
후에 필자는 ubuntu 즉 devian 계열의 운영체제를 사용하고 있으므로 apt를 이용해 redis를 install 해주겠다.
아래의 커맨드를 순차적으로 입력하자.
sudo apt update
sudo apt install redis-server
잘 설치가 되었다.
그리고 redis.conf 파일에서 supervised no 를 systemd로 바꿔주자.
sudo systemctl restart redis.service
를 입력하여 환경을 바꾼 것이 적용되게 해준 후 상태를 확인해보자.
sudo systemctl status redis.service
위와 같이 레디스 서버가 정상적으로 작동하고 있다.
이제 airflow의 conf 파일만 수정해주면 된다.
우리는 status에서 resdis의 주소를 확인해 두었다. 127.0.0.1:6379 !!
즉 로컬 호스트인데 포트는 6379만 뚫어서 쓰고 있다.
해당 url을 broker url에 등록시켜주자.
worker 들이 일하고 난 뒤 결과 보고를 어디에 할지도 정해줘야 한다. (위의 실행 순서도를 참고하면 좋다.)
result backend를 아래와 같이 만들어 준다.
result_backend = db+postgresql://postgres:postgres@localhost/postgres
참조 : https://docs.celeryq.dev/en/latest/userguide/configuration.html#task-result-backend-settings
이제 거의다 왔다!!!!
아직 우리는 redis와 어떻게 airflow와 소통할지 정하지 않았다. 때문에 하나만 더 설치해주자.
pip install 'apache-airflow[celery]' --constraint "https://gist.githubusercontent.com/marclamberti/742efaef5b2d94f44666b0aec020be7c/raw/21c88601337250b6fd93f1adceb55282fb07b7ed/constraint.txt"
airflow celery flower 를 통해 flower를 실행시켜 celery worker node들을 확인할 수 있다.
5555 포트를 이용하고 있기 때문에 localhost:5555 로 접속하면 된다.
airflow celery worker 명령을 통해 우리 컴퓨터를 worker node로 만들어주자.
flower 에서 확인한 모습
내 로컬 머신(worker node) 가 정상적으로 해당 프로세스를 끝마쳤다.
마무리
우리는 이제 celery executor에서 성공적으로 worker node를 추가시켰다.
각각의 컴퓨터 즉 각각 머신에 위에서 한 행동들을 반복하면서 worker node를 n개 이상 늘릴 수 있다. 디테일한 것들은 제쳐두고 (worker node에 task limit 걸기 등등) 실제 서비스에서 사용될 수 있는 celery executor를 운용할 수 있게 되었으면 좋겠다.
