Airflow를 나눠봤습니다. 그런데 이제 Autoscaling을 곁들인

Airflow Clusterization

Airflow를 나눠봤습니다. 그런데 이제 Autoscaling을 곁들인

Airflow의 구성요소들을 다 쪼개볼 겁니다. 어떻게 나눌 것이냐면, 크게 Airflow Main, Airflow DB, Airflow Worker입니다. Airflow Main에는 웹 서버와 스케쥴러를 돌아가게 만들 것이고, DB에는 MySQL을 띄워 놓을 것입니다. Worker는 오토스케일링을 걸어둘 것이구요. 귀찮게 왜 이렇게 하냐구요?

기존에 Airflow는 하나의 서버에서 잘 돌아가고 있었습니다. 그래서 Airflow Basic 글을 통해서 정리해 놓기도 했습니다. 문제는 처리할 DAG와 Task가 너무 많아진 것이었습니다. Airflow의 구성요소 각각이 어떤 역할을 하는지 궁금하시거나 잘 기억이 안나시는 분은 이 글을 읽어보시면 좋습니다.

처리할 작업이 많아지면 Worker 하나로는 처리하기 부담스러워집니다. 스케쥴러가 내려준 많은 작업을 처리하다가 다른 작업을 못하게 현상이 자주 발생하게 됩니다. 그래서! 워커를 늘려줘야겠다는 생각을 한 것입니다. 워커에 오토스케일링을 걸어서, 작업 부하가 걸리면 워커를 늘려서 처리하고, 부하가 줄어들면 워커를 줄이려는 것입니다.

---

작업 계획은 다음과 같습니다.

1. Airflow Main 세팅
2. Airflow DB 세팅
3. Airflow Worker 기본 이미지 세팅
4. Airflow Worker 세팅
5. Assemble!

---

Airflow Main

Airflow Main에는 Webserver와 Scheduler를 구성해 놓을 것입니다. Airflow Main의 기본적인 설정은 이 글에서의 설정과 같습니다. 여기서 MySQL을 따로 가져갈 것이니 MySQL설정만 빼놓고 따라하시면 좋을 것 같습니다. 다만 저는 Airflow 1.10.14 버전을 사용하도록 하겠습니다. 최신 2.0이상 버전은 DAG가 조금 달라져서 수정을 해줘야 하거든요.

Airflow Home

먼저 저는 AWS환경을 이용해서 구축을 할 것이기 때문에 EC2를 하나 만들어주고, 새로 생긴 인스턴스에 Airflow home을 잡아주도록 하겠습니다.

export AIRFLOW_HOME=/home/ec2-user/airflow #기본주소

이 HOME주소는 꽤나 중요합니다. Airflow가 시작되는 곳이기도 하고, 설정 파일을 불러오는 곳이기도 하기 때문입니다. 원하는 주소로 작성하시되, Worker에도 동일한 주소를 입력해주셔야 합니다. 그렇지 않으면 execute_command encountered a CalledProcessError ,Celery command Failed 를 만나시게 될 것입니다. ¹

Airflow의 Dag를 실행하려면 airflow를 구성하고 있는 모든 서버에 동일한 경로와 이름으로 DAG파일이 존재해야 한다는 것을 기억해주세요.

Redis 설정

레디스 설정은 다행히 저번 글에 나와 있는 것과 똑같습니다. 그대로 따라하시면 됩니다.

Airflow 설치

sudo yum update -y

sudo yum install group "Development tools" -y

sudo yum install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel python3-devel.x86_64 cyrus-sasl-devel.x86_64 -y

sudo yum install libevent-devel -y

sudo pip3 install apache-airflow==1.10.14

airflow version #버전이 뜬다면 성공!

설치가 되었으면 airflow initdb , airflow webserver를 사용해서 airflow가 뜨는지 확인해 봅니다.

에러가 발생한다구요? 그렇다면 아래 Trouble Shooting 부분을 참고해보세요. (sqlalchemy version 조정) ²

---

MySQL 설정

Airflow Main 세팅이 끝났다면, DB로 사용할 새 인스턴스를 만들고 MySQL을 설치해줍니다.

이전 글과 달라지는 부분이 DB 세팅에서 발생합니다. mysql 5.7버전으로 설치하는 것 까지는 동일합니다.

sudo wget https://dev.mysql.com/get/mysql57-community-release-el7-11.noarch.rpm
sudo yum localinstall mysql57-community-release-el7-11.noarch.rpm 
sudo yum install mysql-community-server

# 설치된 이후
sudo service mysqld start

airflow 사용자를 추가해봅시다. 먼저 root로 접속을 합니다.

$ mysql -u root -p
# 패스워드를 입력
Enter Password : #/var/log/mysqld.log에 있는 임시비밀번호 사용

달라지는 부분이 발생하는 곳이 여기입니다. airflow 사용자를 생성할 때 모든 접속 % 에 대해서 허용해주어야 합니다. 기존에는 내부 주소, localhost에 대해서 허용했습니다.

# 사용자 생성
create user 'airflow'@'%' identified by '비밀번호';

# DB 권한 부여
grant all privileges on *.* to 'airflow'@'%';

flush privileges;

DB가 설정되었으면 Airflow Main과 연결을 해주어야 합니다. 이 DB 서버가 사용하는 Private IP가 있습니다. 이 주소를 AWS에서 확인하고 Airflow Main의 airflow.cfg에 넣어줘서 어떤 DB를 바라볼지 Airflow Main에게 알려줘야 합니다.

기본적인 airflow main의 airflow.cfg 설정은 이렇게 해줍니다.

# 사용할 dag 폴더 지정
# subfolder in a code repository. This path must be absolute. 꼭 절대경로!
dags_folder = /home/ec2-user/airflow/dags #원하는 위치와 디렉토리로 지정

# executor = SequentialExecutor
executor = CeleryExecutor #원하는 Executor 선택

# sql_alchemy_conn = sqlite:////home/airflow/airflow/airflow.db
sql_alchemy_conn =  mysql+pymysql://airflow:@[mysql서버PrivateIP]:3306/airflow

# catchup_by_default = True
catchup_by_default = False

# broker_url = sqla+mysql://airflow:airflow@127.0.0.1:3306/airflow
broker_url = redis://airflow@[main서버PrivateIP]:6379/0

# result_backend = db+mysql://airflow:airflow@localhost:3306/airflow
result_backend = db+mysql://airflow:비밀번호@[mysql서버PrivateIP]:3306/airflow

# load_examples = True
load_examples = False #예시를 보고 싶다면 True

broker_url , cluster_address에는 Main의 Private IP주소를 넣고, sql_alchemy_conn, result_backend 에는 DB의 Private IP주소를 넣습니다.

이렇게 설정해주고 다시 한번 airflow initdb 를 해주면 새로 작성해준 주소로 DB경로가 입력이 됩니다.

추가로, MySQL서버에도 Redis를 설치해주어야 합니다. broker인 Redis를 통해 scheduler와 executor를 연결해주어야 하기 때문입니다.

위 그림에서 보듯이 main은 webserver와 scheduler를 담당하고 MySQL 서버는 DB와 executor쪽을 담당하게 됩니다. executor에 보낼 매개가 필요하기 때문에 브로커인 Redis를 설정해줍니다.

RBAC

webserver를 띄우고 잘 되는지 확인해봅시다. 잘 나오면 다행입니다. 그런데 1.10.10 이상 버전으로 오면서 UI에 변화된 부분이 있습니다. 아주 좋은 기능이라고 생각하는 것 중 하나인데, 그건 시간대를 드디어 설정해서 볼 수 있다는 것입니다. 기존에는 UTC가 고정이어서 +9해서 계산하는게 영 별로였거든요.

사진에서 보다시피 KST로 설정할 수 있습니다. 그리고 유저도 설정해서 볼 수 있죠. 이걸 적용하기 위해서는 config에서 하나를 더 수정해주어야 합니다.

다시 airflow.cfg로 들어가서

# Use FAB-based webserver with RBAC feature
rbac=True

이렇게 설정해줍니다. RBAC는 Role-Based Access Control의 약자로써, 아까 보셨듯이 유저별로 접근을 통제하는 시스템입니다. 그렇다면 유저도 만들어주어야겠죠.

Airflow Main의 쉘로 들어가서

airflow create_user -r Admin -u admin -e email주소 -f admin -l user -p 비밀번호

이렇게 입력을 해줍시다. 각 플래그의 의미는 다음과 같습니다.

• -r : Role, 역할. Admin, Op, User, Viewer, Public 이 정해져 있고, 커스텀 롤 생성 가능. (자세한 내용은 여기를 참고하세요)
• -u : User명
• -e : Email 주소
• -f : First Name
• -l : Last Name
• -p : Password, 비밀번호

이제 webserver를 열어주고 들어가면

아름다운 원 운동을 바라보면서 멘탈이 흔들릴 수 있습니다. 하지만 이렇게 나오게 된 것은 한 가지를 빼먹었기 때문입니다. 앞서서 우리가 중요한 config를 바꿨을 때는 DB에 알려준 것을 기억하실 것입니다. RBAC는 보안에 관련된 것이니까 중요하다고 볼 수 있겠습니다. 그렇다면 DB에도 뭐가 바뀌었는지 알려주어야 합니다.

airflow upgradedb

이 명령어는 DB를 초기화 시키지 않고 설정값을 DB에 업데이트 시켜주는 역할을 합니다. 이 명령어를 입력해주면?

짠! 1.10.14의 UI가 등장했습니다. 오른쪽 상단의 시간 설정이 가능하다면 성공하신 것입니다.

Fernet Key

RBAC를 통해 보안이 강화되었습니다. 이를 통해 활성화 된 것이 하나 더 있다면, fernet_key 를 통한 encryption입니다. Variable이나 Connection을 이용하는 분이라면, DAG를 실행시켰을 때 JSONDecodeError: Expecting value: line 1 column 1 이런 에러를 마주할 가능성이 높습니다. 실제 에러가 난 부분을 보면 Variable의 Value를 가져오는 부분에서 문제가 발생하고 있습니다. 실제로 어떤 Value를 가져오는지 확인해보면, “g8dgasv90s8fd09x9adxfcx” 같이 알 수 없는 암호문으로 되어 있는 것을 목격할 수 있습니다.

‘아 그렇다면 encrypt 옵션을 False로 바꿔야지!’라는 생각으로 mysql로 들어가서 variables를 찾은 뒤에 is_encrypt 를 다 0으로 변경해봤습니다. 하지만 이렇게 해도 이미 Variable에 등록할 때에 암호화된 코드로 DB에 들어가기 때문에 key값으로 value를 조회해도 나오는 값은 위에 있는 알 수 없는 암호문입니다.

결국에는 이것을 해독해주어야 합니다. 이 암호를 해독해주는 열쇠가 바로 fernet_key 입니다. fernet_key는 한번 생성해주어야 합니다.

from cryptography.fernet import Fernet
fernet_key= Fernet.generate_key()
print(fernet_key.decode())

여기서 나온 키 값을 복사해서 airflow.cfg의 fernet_key 부분에 넣어줍니다. 그리고 이 키 값은 저장해놨다가 airflow worker에 있는 airflow.cfg에도 동일하게 적용을 해줍니다.

---

Airflow Worker 기본 이미지 세팅

Airflow Worker를 세팅해보겠습니다. 그 전에 기본 이미지를 설정해 줄 것입니다. 물론 Worker를 쫙 설치하고 세팅해준다음에 AMI를 만들어서 오토스케일링을 진행해도 되지만, 만약 설정값을 바꿀일이 생긴다면? 그때마다 AMI를 새로 만들어야 할 것입니다. 당연히 AWS관리자나 TA분과의 관계가 좋지 않아질 것입니다.

제가 선택한 방법은 Docker입니다. Docker를 이용해서 위에서 진행한 세팅을 한번에 잡아줄 것이고, 이 이미지를 시작 템플릿으로 설정할 것입니다. 설정 값을 바꿀 일이 있다면 이미지가 저장된 곳에 들어가서 변경한 후 다시 Push해주면 됩니다.

기본 이미지는 Docker와 기타 사용할 명령어에 대한 라이브러리가 설치된 정도면 됩니다. 저는 Docker정도만 설치했고 이것을 AMI로 만들어 줬습니다. 이 AMI를 시작템플릿에 넣을 것이고 시작템플릿에 있는 고급 설정을 통해 서버가 시작되면서 Docker Image를 Run 해 줄 수 있는 명령어를 넣어줄 것입니다.

Docker Airflow Worker 이미지 구성

본격적으로 Worker 이미지를 만들어보겠습니다.

구성요소는 다음과 같습니다.

• Dockerfile
• files
  • airflow.cfg
  • config
    • log_config.py
• sources
  • airflow.sh
  • cron
  • hostname_resolver.py
  • requirements.txt

디렉토리명은 임의로 정해놓은 것이기 때문에 다르게 설정하셔도 됩니다. files부터 설명드리면, airflow.cfg는 위에서 보셨던 그 설정파일입니다. airflow main에서 설정했던 파일을 복사해서 넣어주시면 좋습니다. config 폴더에는 airflow의 log설정과 관련된 파일이 있습니다.

sources에는 airflow.sh로 쉘 스크립트를 통해 실행하는 명령이 담겨 있고, cron작업을 위한 cron, 도커 내부에서 동작하기 때문에 도커의 호스트를 알려주는 hostname_resolver.py, 그리고 필요한 python 라이브러리 설치를 위한 requirements.txt가 있습니다.

Dockerfile

FROM ubuntu:18.04
RUN apt-get update && apt-get install -y \
  git \
  cron \
  vim \
  python3-pip \
  python3-dev \
  build-essential \
  libmysqlclient-dev \
  libssl-dev \
  libkrb5-dev \
  libsasl2-dev \
&& rm -rf /var/lib/apt/lists/*

ADD ./source/* /app/
ADD ./files/airflow.cfg /app/
ADD ./files/config/ /data/airflow/config/
ADD ./source/hostname_resolver.py /usr/local/lib/python3.6/dist-packages/airflow/
RUN chmod 0744 /app/airflow.sh
RUN export AIRFLOW__CORE__HOSTNAME_CALLABLE=airflow.hostname_resolver:resolve
RUN export AIRFLOW__CORE__FERNET_KEY=***
ADD ./source/cron /etc/cron.d/
RUN chmod 0744 /etc/cron.d/cron
RUN crontab /etc/cron.d/cron
RUN touch /var/log/cron.log

WORKDIR /app
RUN mkdir -p /data/airflow/DAGS #제가 만든 DAG 폴더
RUN git clone DAGS #제 github에 있는 DAGs

RUN pip3 install -r requirements.txt
RUN python3 -m pip install sqlalchemy==1.3.15 #에러 방지
#CMD ["tail","-f","/var/log/cron.log"]
CMD ./airflow.sh $NAME

기본 세팅으로 들어가는 라이브러리가 좀 있습니다. git이나 cron, vim등은 자주 사용하기 때문입니다. 그 외에 ADD 부분을 보시면 /source라든가 /files가 있는데 이것은 제가 빌드하기 전에 만들어 놓은 디렉토리 입니다. 이 디렉토리 안에 필요한 파일들을 넣어놨고, 빌드 후에 원하는 위치로 옮기고 실행해줄 것입니다.

export로 환경 설정 해주는 부분이 있는데 여기에 원래는 AIRFLOW_HOME 설정도 해줬습니다. 하지만 이것은 docker run할 때 넣어주기 때문에 빼놨습니다. 그리고 fernet key부분은 아래에서 설명하겠습니다.

airflow.sh

#!/bin/sh
export C_FORCE_ROOT="true"
crontab /etc/cron.d/cron
/usr/local/bin/airflow initdb
mv /app/airflow.cfg /data/airflow/
/usr/local/bin/airflow worker -q $1 &
cron && tail -f /var/log/cron.log

쉘 스크립트에는 Celery 작업을 위한 C_FORCE_ROOT가 있습니다. 이것은 root 권한으로 셀러리를 돌리겠다는 표시이구요. 그 외에는 cron 등록, db initialize 등이 있는데 worker -q에 $1이 있는 이유는 docker를 run할 때 어떤 큐 값으로 실행시킬 건지 정하기 위해서 입니다. 워커를 여러 대 둘 수 있는데 큐를 따로 관리할 수 있으면 좋으니까요.

cron

* * * * * cd /data/airflow/DAGs && git fetch --all && git reset --hard origin/master
* * * * * find /data/airflow/logs/* -ctime +7 -exec rm -f {} \;

cron 작업에는 사용할 DAG들이 있는 git repo를 받는 부분과, log관리하는 코드가 같이 들어있습니다.

hostname_resolver.py

import os
import requests
def resolve():
    return requests.get('http://169.254.169.254/latest/meta-data/local-ipv4').text

hostname_resolver.py에는 aws의 메타 데이터를 이용해서 docker의 host명을 가져오는 코드가 들어있습니다. 혹시 웹 서버에서 DAG 작업로그를 보려고 하는데 나오지 않는다면, 이 부분을 의심해 보세요.

requirements.txt

boto3==1.12.0
pymongo==3.10.1
celery==4.4.0
mysqlclient==1.4.6
pymysql==0.9.3
redis==3.4.1
apache-airflow==1.10.14

requirements에는 필요한 라이브러리들이 담겨있습니다.

이렇게 워커 이미지를 구성했으면, 빌드를 시작해봅시다.

Docker Build and Run

만든 이미지를 AWS ECR에 올려볼 것입니다. 이를 위해 다음과 같은 순서의 작업을 수행하여야 합니다.

1. 인증 토큰을 검색하고 레지스트리에 대해 Docker 클라이언트를 인증합니다. AWS CLI 사용:

   aws ecr get-login-password --region ap-northeast-2 | docker login --username AWS --password-stdin *******

   참고: AWS CLI를 사용하는 중 오류가 발생하면 최신 버전의 AWS CLI 및 Docker가 설치되어 있는지 확인하십시오.

2. 다음 명령을 사용하여 도커 이미지를 빌드합니다. 도커 파일을 처음부터 새로 빌드하는 방법에 대한 자세한 내용은 여기 지침을 참조하십시오. 이미지를 이미 빌드한 경우에는 이 단계를 건너뛸 수 있습니다.

   docker build -t 이미지명 .

3. 빌드가 완료되면 이미지에 태그를 지정하여 이 리포지토리에 푸시할 수 있습니다.

   docker tag 이미지명:tag 리포지토리주소/이미지명:tag

4. 다음 명령을 실행하여 이 이미지를 새로 생성한 AWS 리포지토리로 푸시합니다.
   docker push 리포지토리주소/이미지명:tag

위에 있는 것들은 ECR의 푸시 명령에 있는 것들이기 때문에 해당 페이지에 있는 코드를 복사해서 넣으시면 됩니다.

빌드가 되었다면, 이미지 구성은 끝입니다. 이제 필요한 것은 시작 템플릿을 방금까지 만든 이미지를 이용해서 구성하고, 이것을 오토스케일링 그룹에 넣어서 워커를 쫙 만들어 주면 됩니다.

Start Template

시작 템플릿 구성

시작 템플릿 이름은 잘 넣어주시면 되고, AMI에는 아까 구성한 Docker AMI를 넣어주도록 합니다. 이 도커 AMI를 기본으로 설치하고 그 위에 도커 이미지를 Run 해 줄 것입니다.

이제 맨 밑에 고급 세부 정보로 들어와서 사용자 데이터를 넣어줍니다.

사용자 데이터는 이 시작 템플릿이 시작될 때 실행할 수 있는 명령어입니다. 여기에 docker run 명령어를 넣을 것입니다.

#!/bin/bash
su - ubuntu -c "aws ecr get-login-password --region ap-northeast-2 | docker login --username AWS --password-stdin *******;docker run -p 8793:8793 -e AIRFLOW_HOME=/data/airflow -e NAME=main -d ***********/작성한이미지명:태그"

shebang을 꼭 넣어주셔야 합니다.

shebang을 넣지 않으면 이 명령어를 인식할 수 없어서 에러가 발생합니다. 한 번 더 강조합니다. shebang 꼭 넣어야 한다고! 빼먹어서 저 처럼 삽질 많이 하지 마십시오. 그 뒷 부분에는 su ubuntu가 있습니다. 이후에 실행할 명령어는 ubuntu유저로 실행해야하기 때문입니다. 그래서 switch user를 해주시고 다음의 명령어를 입력해줍니다.

하나씩 자세히 볼까요?

aws ecr get-login-password --region ap-northeast-2 | docker login --username AWS --password-stdin *******

이것은 아까 사용했던 명령어로, 인증 토큰을 검색하고 레지스트리에 대해 Docker 클라이언트를 인증하는 것입니다.

docker run -p 8793:8793 -e AIRFLOW_HOME=/data/airflow -e NAME=main -d ***********/작성한이미지명:태그

도커 이미지를 run하는 명령어입니다. 8793포트를 열어줘야 워커에 대해서 통신이 가능합니다. -p를 이용해 포트를 적어주세요. -e 명령어는 도커에서 환경설정 관련된 부분입니다. airflow home을 airflow main과 같은 위치로 잡아주시면 됩니다. NAME은 worker -q 다음에 들어갈 큐의 topic 명입니다. 제가 작성한 DAG들은 main 토픽을 사용하므로 main으로 했습니다. 다른 topic을 사용하시는 분은 다른 걸로 넣어주세요. -d는 백그라운드로 실행시킨다는 의미이며, 그 다음에는 ECR 이미지의 주소를 넣어줍니다.

시작템플릿 작성이 끝났습니다.

---

Assemble!

이제 구성이 모두 되었습니다. airflow main에 들어가서 airflow webserver와 airflow scheduler를 올려줍니다.

nohup airflow webserver &
nohup airflow scheduler &

오토스케일링 그룹에도 워커를 늘려주고 도커 이미지가 올라올 때까지 조금 기다려줍니다. 다 올라오고 나면 dag를 잘 돌리는지 test 해 봅니다.

airflow list_tasks [TASK명]
airflow [TASK명] [DAG id] [Task id] [date]

잘 돌았다면 성공입니다!

---

Trouble Shooting

1. DAG가 무한 생성?

갓 만든 따끈따끈한 DAG를 On했습니다. 그런데 DAG하나가 끝나기도 전에 새로운 DAG들이 계속해서 실행되는 현상을 목격할 수 있습니다. 이런 경우는 catch_up 옵션이 활성화 되어있기 때문입니다. catupup은 start_date부터 현재 시간까지 실행하지 못한 DAG들을 실행하겠다는 의미입니다. 임시로 실행하는 DAG야 신경 안써도 큰 문제는 되지 않겠지만, API가 엮어있거나 대용량 데이터를 끌어서 사용하는 DAG라면 문제가 커질 수 있습니다. 이는 airflow.cfg에서 해결할 수 있습니다.

catchup_by_default = False

airflow.cfg에서 catchup_by_default 옵션을 찾아서 False로 변경해주시면 됩니다. 기본값은 True입니다.

혹시 이렇게 했는데도 문제가 계속 발생한다면, 웹 서버와 스케쥴러를 모두 종료하신 후에 다시 실행시켜 수정한 옵션 값을 적용해주면 됩니다.

2. pip3 permission denied

열심히 글을 따라 치는데 뜬금없이 pip3 permission denied 이런 에러가 발생할 수 있습니다. 이런 경우에는

python3 -m pip install [library]

# 위 명령어가 안된다면
python3 -m pip install --user [library]

위와 같은 명령어로 해결할 수 있습니다.

3. Command python setup.py egg_info failed with error code 1

역시나 ec2-user에서 열심히 설치를 하는 중에 이 에러를 마주칠 수 있습니다. 이 에러는 pip 업데이트가 되지 않아 발생한 에러로 볼 수 있습니다.

sudo -H pip install --upgrade --ignore-installed pip setuptools

# 또는
python3 -m pip install -U pip

4. ModuleNotFoundError: No module named ‘sqlalchemy.ext.declarative.clsregistry’

2
위의 에러는 sqlalchemy 버전이 맞지 않아서 발생하는 문제입니다. 그렇다면 버전을 낮춰주면 됩니다.

#sqlalchemy 버전 낮추면서 해결
python3 -m pip install sqlalchemy==1.3.15

---

Reference

• 1 출처 : https://louisdev.tistory.com/26
• https://yahwang.github.io/posts/87
• https://potensj.tistory.com/73
• https://oboki.net/workspace/data-engineering/airflow/rbac/
• https://github.com/puckel/docker-airflow/issues/387
• https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/fernet.html#rotating-encryption-keys
• https://devlog.jwgo.kr/2019/07/05/celery-daemonization/
• https://aws.amazon.com/ko/blogs/korea/build-end-to-end-machine-learning-workflows-with-amazon-sagemaker-and-apache-airflow/