[if(kakao) 2021] 추천 시스템 airflow 2.0 도입기 Review

출처 : https://if.kakao.com/session/29

if(kakao)2021

추천시스템 내 workflow 개선을 위해 airflow를 도입했던 이유와, 쿠버네티스에서 airflow를 안정적으로 운영하기 위해 했던 고민들, 그리고 2.0으로의 버전업 경험과 추천시스템 내에서 이를 어떻게 활용하고 있는지 공유합니다.

 

리뷰 포인트

• Airflow 1.10 도입 계기
• 기존 구조의 문제점과 Airflow 2.0으로 업그레이드를 고민한 이유
• 카카오 추천시스템 내 airflow 활용 (2021년 기준)

 

Airflow 1.10 도입 계기

• 문제점
  • 복잡해지는 추천 시스템 파이프라인
  • ML 모델 개발, 배포 난이도에 비해 ML 시스템을 유지하고 관리하는 비용이 매우 크다.
    • ML을 서비스에 활용하기 위해 ML 모델 개발, 배포 외에 필요한 작업들
      • configuration
      • Data Collection
      • Feature Extraction
      • Data Verification
      • Process Management
      • Resource Management
      • Data Analysis
      • Monitoring
      • Serving
  • 늘어나는 트래픽과 서비스
  • 기존에 사용하던 Jenkins 파이프라인으로는 복잡한 Workflow를 관리하는 것에는 한계가 있음
• 어떤 Workflow 관리 시스템을 사용할 것인가? (21년 기준 298 개의 Workflow 관리 시스템이 있음)
  • spotify LUigi - 사례가 부족함
  • Apache Airflow
  • Apache oozie - Hadoop 환경에만 적합함
  • CNCF(Cloud Native Computing Foundation) argo
• 고려 사항
  • Hadoop 환경의 workflow 개선에 도움이 되어야 한다
  • Kubernetes에서 안정적으로 운영할 수 있는 시스템 이어야한다
  • 팀 개발환경 (파이썬 친화적, 컨테이너 기반) 과의 접목
• Airflow VS. Argo Workflows
  • 스케줄러
    • Airflow : 자체 구현 Scheduler
    • Argo : kube-scheduler (쿠버네티스의 스케줄러)
  • 메타데이터 저장소
    • Airflow : PostgreSQL, MySQL
    • Argo : etcd (쿠버네티스의 저장소)
  • DAG 작성 언어
    • Airflow : Python 
    • Argo : YAML
  • Task 간 모델/데이터 전달
    • Airflow : BaseXCom(SQL) 통해 Task 간 직렬화된 객체나 JSON을 공유할 수 있으나 큰 데이터는 전송 못함 또는 직접 구현 
    • Argo : Artifactory, HDFS, S3 compatible, etc. 활용 가능
  • 3rd Party Plugin
    • Airflow : airflow 패키지 내 다수의 플러그인 지원 
    • Argo : 순수 컨테이너 기반
  • Github Start (21/08/25 기준)
    • Airflow : 22,889 
    • Argo : 9,122
• 어떤 Executor를 사용할까? (KubernetesExecutor를 선택한 이유)
  • 실제 프로덕션용으로 사용될 수 있는 Executor는 CeleryExecutor와 KubernetesExecutor 정도가 있다. 
  • CeleryExecutor를 사용하면 Message Broker를 따로 관리해주어야 하는 것이 부담스럽다.
  • 팀 자원 배포환경이 Kubernetes 기반이기 때문에 팀 컨벤션(규칙) 자원 활용 관점에서 KubernetesExecutor를 사용하는 것이 더 이점이 있을 것이라고 생각했다. 
  • Task마다 worker Pod가 생성되어 오버해드가 발생할 수 있으나 대부분 실행하는 Task들의 수행시간이 길기 때문에 큰 문제가 되지 않을 것이다.

 

 

[그림 1] 기존 파이프라인

[그림 2] 새로운 파이프라인

기존 구조의 문제점과 Airflow 2.0으로 업그레이드를 고민한 이유

• 기존 구조 (그림 1)
  • 용어
    • Pod 구성 패턴 (참조)
    • Helm (참조)
  • Airflow를 사용하기 위해 구성해 주어야 하는 것
    • Scheduler
    • Webserver
    • MetaDB
    • DAG file
  • Helm을 이용한 배포 관리
  • 가상머신 서버에 MySQL과 NFS-Server를 띄워 각각 Meta DB와 Log 저장소로 사용
  • 사이드카(sidecar) Pod 구성 패턴으로 보임
    • 먼저 git-sync 컨테이너가 DAG file을 git repository에서 다운로드
    • 사이드카로 붙어 있는 scheduler 컨테이너와 webserver 컨테이너가 Meta DB 인 Mysql 서버와 통신하고, 공유 볼륨(volume, 파일 시스템 상 특정 공간)인 emptyDir에서 DAG file을 읽어서 필요한 프로세스를 수행
    • Scheduler 컨테이너의 Kubernetes Executor가 쿠버네티스 클러스터에 새로운 Worker Pod를 띄움
    • Worker Pod는 KubernetesPodOperator를 실행시켜 Task를 수행할 Pod를 띄우고 이 과정에서 발생하는  Log(Scheduler, Webserver, Task Log)를 NFS-Server에 저장(Logging) 
• 기존 구조의 문제점과 해결방법
  • 유연하지 못한 Worker Pod 스펙 설정
    • 기존 구조의 Airflow 버전에서는 Worker Pod의 스펙(자원, 컨테이너 이미지, 등)을 설정하기 위해 airflow 설정 파일이나 DAG 파일을 작성하면 Airflow 자체 템블릿이 쿠버네티스 API를 사용해 Worker Pod의 스펙을 정의한다. 이는 쿠버네티스 API를 직접 사용하는 것보다 유연성이 떨어지고, 결국에는 쿠버네티스 팟 템플릿에 대한 전반적인 이해가 필요하여 작성이 어렵다. -> 2.0부터는 pod_template_file yaml로 더 직관적이고 유연하게 Worker pod의 스펙을 설정할 수 있다.
  • SPoF(단일 장애지점)
    • 기존 구조에서 scheduler는 하나만 존재하기 때문에 scheduler에 장애가 발생하는 경우 복구되기 전까지는 전체 장애가 발생한다. -> 2.0부터는 두 개 이상의 멀티플 scheduler를 사용할 수 있어 고가용성 구조로 구성 가능하다.
  • DAG가 많아질 수록 로그 작성량의 증가로 자체 구축한 NFS 서버의 퍼포먼스 저하 -> 카카오의 로그 시스템(KEMI)와 연동
    
    
  • scheduler와 webserver가 2.0부터 고가용성 구성으로 구조가 바뀌면서 DAG를 읽기 위해공유 볼륨인 emptyDir을 사용할 수 없다. -> DAG Serialization, Scheduler가 Meta DB에 시리얼화된 DAG를 저장
• 새로운 구조
  • Airflow를 위한 Airflow Pod 뿐만 아니라 Meta DB를 위한 Postgressql Pod, KEMI와 연동된 로깅을 위한 Kocoon-Hermes Pod, 모니터링을 위한 Kocoon-Prometheus Pod를 함께 띄운다. 각 Pod의 자원을 helm chart로 관리
  • 구조 
    • Airflow Pod(main)
      • Scheduler와 Worker에 git-sync를 사이드카로 붙어서 DAG를 가져옴
      • Scheduler와 webserver는 replica 값이 2인 ReplicaSet으로 구성 (airflow 2.0의 멀티플 구성)하고 안티어피니티(AntiAffinity)를 적용해 같은 노드에 여러 개의 Scheduler나 webserver가 배포되지 않도록 하여 노드 장애 발생시 하나의 Scheduler나 webserver에만 장애가 발생하도록 함 
      • Airflow 관련 설정 파일은 쿠버네티스 ConfigMap 오브젝트로 관리하고 scheduler나 webserver와 마운트하여 변경이 발생하면 재실행
    • Postgressql Pod
      • Airflow MetaDB 용으로 postgresql DB 사용 (사내 Postgressql 셀에서 HA구성이 가능한 차트를 제공받을 수 있고, airflow에서 Postgressql DB 이슈가 가장 메인으로 지원됨, mysql보다 이슈 발생 빈도가 낮음)
    • Kocoon-Hermes Pod
      • fluentd가 모든 노드에 떠있으면서 Worker, scheduler, webserver의 로그를 카카오 로깅 시스템인 KEMI로 전송
    • Kocoon-Prometheus Pod
      • 카카오 클라우드 텔레메트리 파트에서 제공되는 올인원 모니터링 차트 

 

카카오 추천시스템 내 airflow 활용 (2021년 기준)

• 데이터 파이프라인 관리
• 컨테이너 구성 - 기반 라이브러리 + Airflow 
• BaseXcom 과 HDFS를 활용한 Task 간 데이터 전송
• ML 파이프라인 - GPU Training
  • CPU연산은 airflow가 동작하는 CPU Kubernetes 클러스터 내에서 진행
  • GPU 연산은 KubernetesPodOperator를 사용해 GPU Kubernetes 클러스터에서 진행

 

마무리

•  workflow management system을 도입하기 위해 고려한 요구사항과 하는 사항과 argo, airflow 등 여러 종류의 기술 스택 중 하나를 선택할 때 어떤 feature를 고려했는지 알 수 있었다. 데이터 엔지니어링에서 어떤 기술 스택을 선택할 때는 해당 기술 스택의 장단점, Tradoff, 아키텍쳐와 동작원리를 고려했을 때 우리의 데이터나 기존 시스템과 잘 호환되는 지 등을 고려해야 한다.  
•  kubernetes위에서 airflow 동작시킬 때, 기존 방식과 새로운 방식을 비교하면서 어떤 Pod를 띄워야 하는 지 알 수 있었다.
• 모든 노드에 fluentd Pod를 띄우면 로그 수집을 할 수 있다는 것을 알 수 있었다. 아마도 해당 노드의 파일 시스템에 있는 log 관련 파일들을 수집하는 것 같다.