수동으로 하던 노가다 작업을 Airflow를 이용해서 자동화하기
들어가며
안녕하세요. 밀당 데이터팀 한승연입니다.
밀당 데이터팀에서는 데이터를 통해 학생의 학습경험을 증진하거나, 밀당 직원들의 업무 효율을 높일 수 있는 방법을 고민하고, 이를 위한 머신러닝/딥러닝 모델을 테스트하고 개발합니다. 데이터팀의 메인 언어가 파이썬이다 보니, 모델을 개발하는 본 업무 외에도, 파이썬으로 개발된 서비스를 유지, 관리하는 사이드 프로젝트 또한 데이터팀에서 맡고 있습니다.
오늘 글의 주제는 데이터팀 (노가다) 사이드 프로그램의 Airflow를 활용한 자동화입니다. Airflow는 흔히 데이터 파이프라인 오케스트레이션 툴로 알려져 있습니다. Airflow가 파이썬 기반으로 태스크를 구성하고 워크플로우를 관리하는 툴이다 보니, 데이터 파이프라인을 관리하는 목적 외에도 이번 파이썬 기반의 프로그램을 자동화 하는데도 사용하게 되었습니다.
왜 airflow가 필요해졌는가?
월간리포트 간단 배경
밀당에서는 매월 학생의 학습 성취 결과를 부모님께 리포트로 전송 해드립니다. 해당 리포트는 학생들의 학습결과와 선생님의 코멘트로 나뉘게 됩니다. 학습결과 계산은 학생 이메일을 사용하여 서버에서 자동으로 계산이 되지만, 선생님의 코멘트는 밀당의 문서관리 툴 CODA에 선생님들께서 직접 입력해주십니다. 리포트는 두 가지 데이터를 이용하여 HTML 렌더링을 통해 탄생됩니다.
*코다참고: https://coda.io/@steve/coda-101
코다 종속성 문제
코다 사용의 문제점은 코다에서 제공하는 API의 속도가 너무 느려 개발환경에 적합하지 않다는 점입니다. 매실행마다 코다 api를 통해 데이터가 수집되다 보니, 데이터의 개수가 늘어갈수록 실행속도가 느려집니다. 약 3000명의 학생에 대해 매월 데이터가 쌓이므로 만건이 넘어가는 데이터를 API를 통해 받아오는데 길게는 3분까지 걸리는 상황입니다.
또한 안그래도 느린 코다 API에 코다 데이터 수집-리포트생성-발송 프로세스가 일률적으로 이루어지다 보니 다양한 문제점이 발생합니다.
코다에서 수집한 데이터로 리포트를 생성하고 문자를 발송하면, 생성된 URL과 문자 발송여부를 코다로 다시 업데이트를 해주어야했습니다. API를 이용하여 하나의 셀을 업데이트하는데 짧게는 3초 길게는 10초가 넘어가기 때문에, 3000명의 학생에 대해 코드가 실행이 되면 프로세스의 시작부터 종료까지 몇시간씩 걸리게 됩니다.
더불어 모든 데이터가 코다에 안전하게 업로드 되기 이전에 프로그램이 다시 실행되었을 때도 문제가 발생합니다. 코다는 API 호출 시점의 데이터를 주기때문에, 이미 발송이 되었지만 코다에 업로드가 완료되지 않은 학생에게는 여러번 재발송되는 경우도 있었습니다.
로컬 실행의 문제
월간리포트 프로그램의 최초 개발은 개발팀이 아닌 기획팀에서 시작되었고, 기획팀의 다른 업무와 병행하여 하루만에 빠르게 스크립트를 작성해야 했습니다. 이로 인해, 개발팀이 주로 사용하는 AWS 서버에서 실행되지 않고 로컬 서버에서 프로그램이 실행되어야 했습니다. 즉, 리포트 생성로직을 담은 API는 Heroku 서버를 이용하여 실행되지만, 학생마다 리포트 생성요청 API를 호출하는 작업은 로컬에서 실행되어야 했습니다.
월간리포트 작업이 개발팀에 넘어오게 되면서, 밀당 수학의 경우 매주, 영어의 경우 매달, 3000명이 넘는 학생에 대해, 요청이 있을 때 마다 팀원이 컴퓨터가 있는 자리로 돌아와서 파이썬 코드 실행버튼을 눌러주고 있는 상황이었습니다.
“ㅇㅇㅇ 학생 리포트 다시 생성해주세요”
“ㅁㅁㅁ 학생 리포트가 생성되지 않았습니다. 확인부탁드립니다”
“☆☆☆ 학생 리포트 재발송 해주실 수 있나요?”
와 같은 질문에 일일히 코드를 실행하고 오류를 확인하느라 시간을 뺏겨, 데이터 팀 메인 프로젝트 진행에까지 영향을 미치게 되었습니다. 이에 월간리포트 생성 및 발송을 자동화하고, 오류가 발생할 때만 슬랙을 통해 확인 가능하도록하여, 최대한 사람의 시간을 적게 들일 수 있도록 하고자 하였습니다.
자동화의 필요성
- 코다로 데이터를 관리하는 방법이 너무 취약했으며, 코드를 통해 데이터를 가져오고, 업데이트하고, 추가하는 작업의 시간이 상당히 길어지고 있었습니다. 이에 데이터는 DB를 통해 관리하고, 코다에 선생님이 입력하는 데이터를 주기적으로 싱크해주기로 하였습니다. 리포트 생성과 발송은 DB를 기준으로 실행되므로 코다와 연동하느라 시간을 뺏기지 않을 수 있습니다.
- 기존의 일률적으로 실행되던 코다 데이터 연동, 리포트 생성, 발송 프로세스는 각각 독립적으로 실행되도록 분리하기로 결정했습니다. 프로세스를 분리하여, 에러가 났을 때 어떤 프로세스에 문제가 생겼는지 또한 빠르게 확인가능 하도록 계획했습니다.
- Airflow 는 파이썬 기반의 스크립트를 지원하며, cron 시간표현식에 따라 task 러닝을 스케줄링할 수 있기 때문에 Airflow에 월간리포트 생성 및 발송 작업을 올려 자동화하기로 결정했습니다.
- 각각의 프로세스는 Airflow DAG으로 세분화 하기로로 했으며, 기존의 Heroku 서버 대신 밀당 개발팀에서 사용하던 AWS EKS 의 Kubernetes 환경을 이용하여 로컬 실행을 탈피하고자 하였습니다.
자동화를 위한 Kubernetes 상의 Airflow 설정 및 Github 연동 과정에서 겪은 시행착오들을 어떻게 해결했는지 등 소개드리려 합니다. 이번 글은 데이터 파이프라인 정립하는 과정이 아닌, 단순 수동 작업을 Airflow를 통해 자동화하는 간단한 과정을 담았습니다.
Airflow 설정
차트 설정
Kubernetes에 apache-airflow 공식 헬름 차트를 이용하여 Airflow-2.3.0.을 새로 install 하였습니다. 다른 차트는 버전 업데이트가 신속하지 않은 취약점이 있어 공식 차트를 채택하게 되었습니다. 차트 릴리즈시 차트의 value.yaml 수정을 통해 git-sync, user-setting 등 커스토마이징을 할 수 있습니다. 프로그램 실행에 가장 중요했던 커스터마이징 세가지를 아래에 소개하겠습니다.
git-sync
Airflow가 어떤 코드를 실행할지 DAG을 통해 전달해주어야 합니다. DAG 코드를 shared volume에 직접 저장해줄 수 있지만, 코드 변경이 있을 때마다 이를 변경해주는 작업이 번거롭기에, helm차트에서 제공하는 git-sync 를 이용합니다. git-sync 사이드카는 git Repository를 바라보며, 해당 폴더를 1분 주기로 가져와 저장합니다. Worker와 scheduler pod은 git-sync 사이드카가 저장해준 git repository 코드의 모든 dag 을 설정해준 실행 주기마다 실행합니다. pod 내에서 repository 코드는 /opt/airflow/dags/repo 에서 확인할 수 있습니다.
Pod Settings
node selector
kubernetes 상에 팟이 생성될 때, 설치될 노드의 리소스를 확인하여 알아서 분배되어 설치됩니다. 기존에 존재하던 노드의 환경이 airflow와 맞지 않아 계속해서 설치에러를 맞닥뜨렸습니다. 이에 새로운 노드를 생성하고, 해당 노드에 에어플로우를 설치하기로 했습니다. value.yaml에서 팟들을 어떤 노드에 생성할지 지정해줄 수 있었습니다.
노드에 유니크한 레이블을 지정해준 후 노드셀렉터를 이용하여, 해당 노드에 팟을 설치하도록 설정해줍니다.
특정 노드그룹으로 지정해줄 때의 주의할 점은 노드그룹의 cpu와 메모리 용량을 잘 따져봐야 한다는 점입니다. 해당 노드에 airflow만 설치되어 있지 않을 것이며, airflow는 task가 러닝 중이지 않을때에도 상당한 양의 메모리를 차지합니다.
월간리포트의 프로그램은 상당한 양의 학생 데이터를 다루기 때문에, 하나의 task 실행이 길게는 5–10분 걸릴 때가 있습니다. 이러한 task가 한번에 여러개 실행될 경우 노드의 용량을 초과하여 팟이 재생성되는 경우가 발생했습니다.
따라서, 특별히 airflow와 충돌하는 노드가 없다면 노드 셀렉터를 굳이 지정하지 않는게 베스트 솔루션일 것 입니다. 굳이 노드를 지정해주어야 한다면, 노드의 사양을 업그레이드 해줄 수 있습니다. 노드 셀렉터에는 한 개 이상의 노드를 지정하여 리소스 부족 시 알아서 재분배 될 수 있도록 설정할 수도 있습니다.
Path 설정
월간리포트 프로그램은 프로그램 내에서 여러 폰트와 html 파일을 불러와서 사용하고 있습니다. 로컬 환경과 github 실행시에는 repository의 root 폴더로부터 절대경로를 지정해주면 되지만, 에어플로우는 repository의 코드를 자체 volume 에 저장해두기 때문에 해당 경로를 찾아가지 못합니다.
에어플로우의 절대경로는 /opt/airflow/ 로부터 시작하며, repository 코드는 /opt/airflow/dags/repo 폴더에서 찾을 수 있기 때문입니다.
에어플로우 차트 릴리즈시에 value.yaml에서 기본 root를 수정하여, root 폴더가 바로 git repository를 바라볼 수 있도록 설정할 수 있습니다.
Library 설치
Airflow는 DAG에 지정된 파이썬 함수를 실행시켜주는데, Airflow에서 제공하지 않는 패키지를 사용할 경우에는 Airflow의 worker, secheduler pod에 직접 설치를 해주어야 원활히 함수를 실행해줍니다.
이 방법의 문제점은 리소스 초과 등의 이유로 팟이 재생성될 경우, 다시 직접 worker, scheduler pod의 bash에서 다시 pip install 해주어야 한다는 것입니다.
따라서, 팟이 재생성될 경우 자동으로 requirements.txt를 install 해주도록 설정하고자 했습니다.
Airflow pod은 apache/airflow docker hub 의 이미지를 이용하여 빌드합니다. 팟이 재생성 될때마다 해당 이미지가 빌드되므로, bash에서 requirements.txt를 install 하도록 커스터마이즈 합니다. value.yaml 에서 밀당의 docker repo에 있는 해당 이미지를 바라보도록 수정해줌으로써 자동 pip install 설정을 할 수 있었습니다.
*참고
charts/extra-python-packages.md at main · airflow-helm/charts
Web UI
web에서는 dag과 task의 상태를 확인할 수 있을 뿐만 아니라, 커맨드에서 할 수 있는 다양한 설정을 간편하게 할 수 있도록 제공하고 있습니다.
LoadBalancer-Ingress
Web UI는 기본적으로 서비스 포트포워딩을 통해 로컬에서 확인할 수 있습니다. 별다른 설정을 하지 않은 이상 서비스는 클러스터 IP만 할당받기 때문입니다.
매번 web UI를 할때, 커맨드를 열어 포트포워딩 해주어야 하는 귀찮음이 있습니다. 또한, 아래에서 다룰 task running 실패시 보내주는 슬랙메세지에 바로 로그를 확인할 수 있는 링크도 한번의 클릭으로 확인할 수 없다는 불편함이 있습니다.
가장 간단한 방법으로는 아래와 같이 webserver의 서비스 타입을 로드밸런서로 지정해 준 후 External-IP를 얻는 방법입니다.
하지만 해당 방법은 서비스가 업데이트 될 때마다 IP가 변경됩니다. 따라서 월간리포트 airflow에서는, IP기반이 아닌 URL 기반으로 경로를 설정하고, 여러개의 로드밸런서를 두지 않고 하나의 URL로 웹 서비스에 접속할 수 있도록 서비스 타입을 노트포트로 해준 뒤 Ingress와 연결해주었습니다.
log 확인
web에서 실행되는 task마다 log를 확인할 수 있습니다.
python operator의 경우, 실행되는 함수의 print문과 return 값을 log에서 확인할 수 있습니다. 파이썬 함수의 적절한 곳에 print문을 배치하여, 함수의 어떤 부분에서 함수가 났고 어떤 데이터에 문제가 있는지를 확인할 수 있도록 했습니다.
Variable 설정
서버에서 직접 파이썬 함수를 실행할 경우, 민감한 key 값 또는 password은 .env 파일을 이용하여 실행하게 됩니다. 하지만, .env와 같은 민감한 정보는 github에 업로드 할 수 없으므로, airflow는 다른 방식으로 정보를 받아와야 합니다. 이 때 airflow에서 제공하는 Variable을 사용할 수 있습니다.
Variable은 web UI의 Admin > Variables에서 간단하게 설정 가능합니다.
이렇게 설정해준 variable은 dag에서 airflow.models.Variable 을 통해 가져올 수 있습니다.
python operator는 op_kwargs를 통해 함수에 인자를 넘겨줍니다. variable.get을 통해 받아온 정보를 op_kwargs로 넘겨준다면, 문제없이 해당 key를 이용하여 함수가 실행됩니다.
*더욱강력한 보안이 필요한 정보에 대해서는 여기를 참고하세요
DAG
Time schedule
cron표현식
Task가 running 되는 주기를 time schedule을 통해 설정할 수 있습니다. 주기설정은 cron 시간표현식을 따릅니다. AWS 상에서 airflow를 사용한다면 cron 표현식이 살짝 다르다는 점을 주의해야 합니다. 기존 표현식에서 요일이 sun-sat : 0–6로 표현되는 반면, AWS 표현식에서 요일은 sun-sat : 1–7로 표현됩니다. aws 시간표현식은 여기서 확인할 수 있습니다.
timezone
한국 시간을 기준으로 start_date와 time schedule을 설정했다면 주의해야 할 점은, 기본 설정은 UTC로 지정된다는 것입니다. 예를들어, 월간리포트 발송은 너무 이르거나 늦은시각에 발송되면 안되므로, 9시부터 20시 사이에 발송되도록 설정을 해두었습니다. 하지만 UTC 기준으로 실행된다면, 실제 발송은 12AM ~ 9AM 사이에 발송되어버립니다. 따라서 DAG 작성시 pendulum 패키지를 사용하여 시간 기준을 KST로 설정해주었습니다.
Task Execution
task는 start_date을 시작으로 schedule_interval 마다 실행됩니다. 만약 start_date를 일주일전으로 하고, 한시간에 한번씩 설정해둔 뒤 오늘 Dag을 활성화 시키면, 일주일 동안 돌지 못했던 task가 168(=24*7)번 동시에 실행되게 됩니다. 이 경우, 고사양의 노드를 사용하지 않는 이상 높은 확률로 노드 메모리를 초과하여 사용하게 되고, 팟이 재생성될 것 입니다.
이미 필요한 패키지가 자동 install 되도록 설정해두었으니, 기다리면 팟이 재생성되기를 기다릴 수 있지만, dag에 옵션을 걸어줌으로써 이를 사전에 방지할 수 있습니다. 두가지 다른 방법이 있습니다.
max_active_runs
max_active_runs는 태스그 한번에 몇개까지 실행될 수 있는지를 지정해줍니다. max_active_runs=1로 지정해두면, 과거 실행되지 못했던 task가 쌓이더라도 한번에 하나씩 실행됩니다. 문제점은 여전히 168개의 task가 한번에 하나씩 실행된다는 점입니다.
그럼에도 max_active_runs를 적절하게 설정해주면 좋은 점은 수행되지 않은 기간의 dag이 동시실행 외에도, dag이 여러번 trigger 되었을 때 노드 메모리를 많이 사용하지 않도록 할 수 있다는 점입니다.
catch_up
catch_up 의 경우, 말 그대로 스케줄되었지만 수행되지 않은 기간의 dag을 실행할 것인가에 대한 여부를 지정해줄 수 있습니다. catch_up=False 로 설정해줄 경우, pause 상태의 dag 을 오랜만에 실행하더라도, 밀린 dag을 다시 실행하지 않도록 할 수 있습니다.
Slack Alarm
Task running이 실패했을 때, 자동 슬랙 메세지가 발송될 수 있도록 설정할 수 있습니다. Airflow에서 제공하는 Slack Operator를 활용할 수 있지만, 버전 호환성 이슈가 종종 발생한다는 의견이 있어 안전하게 파이썬 패키지 slack_sdk 를 이용하기로 했습니다.
slack_sdk를 이용하여 메세지를 인자로 받아, 슬랙봇이 지정된 채널로 메세지를 전송할 수 있도록 함수를 만듭니다. DAG에서는 이 함수를 호출하여 fail 시 해당 함수를 실행할 수 있도록 설정했습니다.
마무리하며
지금까지 간단하다면 간단한 airflow를 이용한 프로세스 자동화를 소개해드렸습니다.
하나의 함수로 작성되어 사람이 일일히 실행해주어야 했던 월간리포트 발송 프로세스가, airflow를 이용하여 자동화될 수 있었습니다. 이제 월간리포트는 데이터의 조건을 확인하여 자동으로 코다와 밀당DB를 연동해주고, 월간리포트를 생성해주고, 검수가 완료되면 학부모님께 발송됩니다. 실무자가 재생성, 재발송 등의 이유로 데이터팀에 요청하지 않아도 조건에 따라 주기적으로 자동 생성 및 발송되므로 비효율적인 커뮤니케이션도 줄어들게 되었습니다. 결과적으로 데이터팀에는 (아주 가끔) task가 실패할 경우, slack 메세지를 받고 log를 확인하여 문제를 수정하는 일만 남게되었습니다.
앞으로 나아가야 할 방향
월간리포트가 데이터팀의 메인 작업이 아니기 때문에 가장 간단한 방식으로 airflow를 사용했지만, 앞으로 더 고도화할 수 있는 작업도 여전히 남아있습니다.
Kubernetes Executor 채택의 필요성 고민
Airflow Executor는 태스크의 실행방식을 결정합니다. 공식 헬름 차트의 기본 설정은 Celery Executor입니다. Kubernetes 상에서 Celery Executor의 단점은 postgre, redis, scheduler, websever, worker 등 다양한 팟들이 노드의 리소스를 꾸준히 차지한다는 점입니다. 일례로 노드셀렉터 부분에서 설명드렸듯이, dag이 실행되거 있지 않아도 노드의 메모리를 꽤 차지하고 있었다는 점입니다.
이에 반해, Kubernetes Executor의 경우, 태스그가 실행될 때 워커가 생성되고 실행 후에는 다시 삭제되므로 자원을 효율적으로 관리할 수 있습니다. 현재 밀당의 에어플로우를 통해 사용하고 있는 DAG의 수가 많지 않고 Kubernetes Executor의 경우 설정이 복잡하여 Celery Executor를 사용하고 있습니다. 하지만 앞으로 더욱 워크플로우가 복잡해질 경우 더 효율적인 오케스트레이션을 위하여 Kubernetes Executor를 채택하는 것에 대한 고민이 필요합니다.
git-sync sider car 에서 branch 관리의 필요성
현재 git-sync는 github repository의 develop branch를 바라보고 있습니다. 코드 수정을 위해 새로운 브랜치를 생성했을 때, airflow상에서의 실행이 잘되는지 log가 잘나오는지 등을 테스트 하기 위해서는 develop branch로 머지해야 합니다. 물론 로컬에서 테스트를 하겠지만, airflow 환경에서의 실행은 또 다른 문제이기 때문입니다.
해당 repository에서 branch를 새로 생성했을 때, 생성된 branch를 테스트할 수 있는 airflow 환경이 배포되고 테스트를 통과하면 develop branch로 머지되는 프로세스가 필요합니다. 현재는 helm chart를 통해 릴리즈 되었기 때문에 CI/CD 설정이 복잡하고, 앞으로의 새로운 서비스가 자동화될 예정은 없기 때문에 데이터팀의 우선순위가 아닙니다. 하지만 앞으로 airflow상에서 더 많은 서비스가 자동화된다면, 위의 airflow와 연동된 branch 관리 방법에 대한 고민이 필요할 것입니다.
Airflow의 활용가능성
밀당은 LMS, CMS, B2C 등 페이지 및 프로세스를 전면 개혁한 새로운 시스템의 배포를 앞두고 있습니다. 초기 스타트업 단계에서 흔히 그렇듯 레거시 시스템에서는 데이터 수집 설계의 부족으로, ML/DL 프로젝트의 진행이 원활하지 않았던 적이 여러번 있었습니다. 새로운 시스템에서는 데이터의 양도 증가할 뿐 아니라, 새로운 종류의 데이터도 수집될 예정입니다.
이에 따라 데이터 팀에서는 새로운 종류의 데이터 파이프라인을 설계해야 합니다.
- 데이터팀의 메인언어가 파이썬이라는 점
- 파이썬을 이용하여, SQL operator, bash operator, postgres operator 등 워크플로우에 맞게 필요한 오퍼레이터를 확장할 수 있다는 점
- 간편하고 강력한 웹 UI를 이용하여 손쉽게 태스크 트리거가 가능하고, 로그, 그래프, 변수 등의 편리한 뷰를 제공한다는 점
- Kubernetes 상에서 Kubernetes Executor를 이용하여 스케일 아웃이 가능하다는 점
등의 장점을 고려했을 때, 데이터 파이프라인 오케스트레이션 툴로 이미 밀당 AWS에 릴리즈가 되어 있는 Airflow를 채택하는 것이 유리할 것으로 보입니다.
또한, 이번 월간리포트 자동화를 진행하던 중, 밀당 학적팀의 요청에 의해 학부모에게 보낼 긴 진단고사 결과링크를 bitly 서비스를 이용하여 짧은 링크로 바꿔주는 간단한 프로세스의 자동화를 함께 진행했습니다. 한번 릴리즈된 airflow에 필요한 파이썬 코드를 짜서 올리는 것은 간단한 일입니다.
밀당의 다른팀에서도 자동화만 진행된다면, 효율적으로 리소스를 분배할 수 있는 간단한 프로세스가 많이 있을것이고, 앞으로 차차 다른팀과의 소통을 통해 회사 업무 효율화를 증대할 수 있을 것이라 기대하고 있습니다.
