Data Engineering/Workflow

1. Airflow on Kubernetes

Kim Jinung 2022. 11. 22. 17:47

Apache Airflow는 쿠버네티스에 친숙한 프로젝트를 목표로 하고 있으며, 공식 helm 차트를 관리하고 있다. 해당 차트를 이용해서 쿠버네티스 클러스터에 비교적 쉽게 에어플로우를 구축할 수 있다.

Goal

- Airflow를 Kubernetes cluster에서 운영할 때의 장점이 무엇인지 파악한다.

- Helm chart를이용해서 Airflow를 K8s 환경에 구축한다.

1. Airflow를 왜 kubernetes에서 운영해야 할까?

Kubernetes의 핵심은 Container를 개발자가 원하는 상태로 유지시켜주는 것이다. 그리고 이를 Airflow에도 적용할 수 있다.

 

Airflow scheduler, webserver, flower 등을 pod로 띄워서 주기적으로 health check 하고 문제가 생기면 다시 pod를 띄워준다.

여기까지가 Container ochestration이 제공하는 기본적인 혜택(?)이다. 

 

Airflow가 작동하는 원리를 이해하면 Airflow를 K8s에서 운영할 때의 더 큰 이점을 알 수 있다.

 

Airflow architecture

https://airflow.apache.org/docs/apache-airflow/stable/concepts/overview.html

 

Airflow에서 scheduler는 실행 할 workflow를 executor에 전달한다. executor는 적합한 worker에 task를 할당 한다. 그런데 Native airflow에서는 위 Worker가 항상 구동 중이여야 한다.  항상 구동 중이여야 한다는 의미는 worker가 지속적으로 리소스를 점유 하고 있어야 한다는 의미다.

 

 

Airflow kubernetes executor

Kubernetes cluster에서 Airflow Kubernetes executor를 사용하는 아래와 같은 아키텍처를 구성하게 된다. 

https://airflow.apache.org/docs/apache-airflow/stable/executor/kubernetes.html

 

구조는 거의 비슷하지만 worker는 항상 리소스를 점유하지 않는다. airflow가 스케쥴링된 DAG를 실행하기 위해는 쿠버네티스에게 worker pod 생성을 요청한다. (이때 pod의 리소스를 제한 할 수도 있다.)이후 생성한 pod(worker)에 task를 할당하고, task를 완료한 후에는 리소스를 다시 반납한다. 즉 유동적으로 리소스를 사용할 수 있다. 

Airflow는 실시간 처리가 아닌 배치 처리를 위한 workflow engine이고 그렇기 때문에 분명 스케쥴이 돌지 않는 시간이 존재한다. 대부분의 운영 환경에서 DAG Job이 도는 시간 보다는 Idle 상태로 worker가 리소스를 점유하고 있는 시간이 더 길 것이라고 생각한다.

 

Airflow kubernetes pod operator

Airflow는 다양한 operator를 제공한다. BashOperator, PythonOpertor, SparkSubmitOperator 등이 있으며 커스텀 오퍼레이터를 개발해서 사용할 수도 있다.

 

쿠버네티스 executor는 동적으로 worker pod를 생성해서 task를 실행한다. 그런데 여기서 라이브러리 종속성 문제가 발생하게 된다. 하나의 쿠버네티스 클러스터 내에서 하나의 에어플로우를 운영할 때 Spark2, 3 버전을 둘 다 사용해야 하는 경우를 예로 들 수 있다.

 

디테일한 내용은 다음 자료를 참조하면 좋다. 

Kubernetes를 이용한 효율적인 데이터 엔지니어링(Airflow on Kubernetes VS Airflow Kubernetes Executor) - 2

 

Airflow는 Kubernets pod operator를 제공한다. 해당 오퍼레이터는 pod 생성 시 유저가 원하는 container image를 선택하여 worker pod를 생성하여 라이브러리 종속성 문제를 해결 할 수 있다. 

 

(아직 라이브러리 종속성 문제를 겪을 정도로 큰 플랫폼을 운영한 경험은 없어서 kubernetes pod operator를 사용해본 경험이 없다. 추후 시간이 된다면 사용 경험을 정리한다.)

 

2. Airflow on Kubernetes

prerequisite

  • MacOS
  • minikube
  • helm

1. Minikube start

우선 minikube를 종료하고 cpu 및 memory 설정을 진행한 후에 minikube를 실행한다.

 

minikube stop

minikube config set cpus 6

minikube config set memory 20480

minikube start

 

 

2. Installing the Airflow helm chart

 

Helm Chart for Apache Airflow — helm-chart Documentation

 

airflow.apache.org

 

helm repo를 추가하고 airflow를 실행한다.

helm repo add apache-airflow https://airflow.apache.org

helm upgrade --install airflow apache-airflow/airflow --namespace airflow --create-namespace

 

 

Installing 후 아래와 같이 airflow가 k8s 상에서 실행 중인 것을 확인할 수 있다. 별도 옵션 설정 없이 실행했으므로 kubernetes executor가 아닌 ceclery executor가 설정되었기 때문에 worker가 함께 떠있다.

kubectl get pod -n airflow

 

3. Airflow webserver port forward

Port forwarding을 통해 Ariflow webserver 포트를 열어준 후에 접속할 수 있다.

 

default username, password는 모두 admin이다.

 

 

 

 

Airflow webserver에 접속하면 아래와 같은 화면을 볼 수 있다.

 

마치며

이제 DAG를 작성하면 Webserver에서 DAG를 주기적으로 실행 할 수 있다. 그런데 helm chart로 띄운 airflow는 kubernetes cluster에 container로 띄운 상태라, 언제 재시작을 해도 이상하지 않다.  DAG 파일을 저장한 pod가 종료되었다가 재시작이 되어도 이상하지 않다. 따라서 별도의 DAG 저장소가 필요하다.

 

k8s에서는 persistence volume을 제공하므로 이를 활용할 수도 있지만 Airflow는 git-sync라는 아주 좋은 기능을 제공한다.