Kubernetes를 이용한 효율적인 데이터 엔지니어링(Airflow on Kubernetes VS Airflow Kubernetes Executor) 영상을 본 후, 정리한 글 입니다.
데이터 엔지니어링이란?
근래에 데이터 관련 직군에서 엔지니어링 업무쪽에 전문적으로 ownership을 갖는 데이터 엔지니어라는 세부적인 직무가 생겨나고 있습니다
데이터의 ETL(Extract, Transform, Load)가 중요해지는 만큼 해당 직무는 데이터를 다루고 data driven하게 업무를 하는 회사에서는 Data Engineering은 간과해서는 안 될 중요한 부분입니다.
실제 서비스를 제공하는 IT 회사에서는 앱 또는 웹으로 사용자에게 서비스를 제공하면서 로그 데이터, 트렌젝션 데이터, 서비스 데이터 등 다양한 데이터가 생성되게 됩니다.
이러한 데이터를 수집해서 데이터 저장소(RDB, Storage 등)에 저장하게 됩니다. 이러한 데이터를 바탕으로 데이터 웨어하우스, 데이터 레이크를 만들고 이를 새로운 형태로 변환하고 분석하고 EDA되는 과정, 다른 플랫폼과의 데이터 전송 등의 모든 업무가 데이터 엔지니어링 입니다. 회사에 따라서 데이터 엔지니어링과 데이터 분석(DA)을 세분화하여 나눌 수 있고, 이를 하나의 DE팀이 담당하여 하는 경우도 있습니다.
이런 과정 속에서 ETL, 리소스, 네트워크 및 인프라 관리, 새로운 데이터 플랫폼 개발, SQL 스킬 및 쿼리 튜닝, MLOps 등 다양한 도메인 지식이 필요하고 이를 전반적으로 공부하기 때문에 개인적으로 백엔드 개발과 데이터 양쪽에 걸쳐있는 직무라고 생각합니다.
이러한 데이터들이 새로운 비즈니스 모델을 제시하거나 의사결정 등을 할때 중요한 지표의 역할이 될 수 있고, 이러한 데이터들을 활용하여 ML, DL, AI의 학습 데이터, 모델링에 활용할 수 있습니다.
데이터 엔지니어가 하는 일에 대해 자세한 내용을 원하시면,
아래 카일님이 잘 정리를 해 주신 영상을 보시면 도움이 됩니다.
https://www.youtube.com/watch?v=1Viu7Ce632s
데이터 엔지니어링을 조금 쉽게 할 수 있도록 도와주는 도구가 바로 Airflow 입니다.
Airflow 란?
워크플로우를 코드로 작성할 수 있고 스케줄링하고 모니터링 하는 플랫폼 입니다.
Airflow의 구성요소
DAG
- 개발자가 코드로 작성하는 워크플로우(흐름)
Task
- 하나의 작업 단위
- 예) 서비스 DB의 데이터를 데이터 웨어하우스로 가져온다.
- Batch
- 특정 트리거 또는 시간에 따라 반복적으로 작업을 수행할 수 있습니다. (Cron Job)
- Real Time
- 실시간으로 실행합니다. (예 - 로그 모니터링 등)
- Batch
Airflow의 동작 원리
출처 : https://engineering.linecorp.com/ko/blog/data-engineering-with-airflow-k8s-1Web Server
- Web UI에 접속하기 위한 서버
Scheduler
- 개발자가 정의한 Python dag 스크립트를 주기적으로 읽습니다.
- 개발자가 정의 해놓은 특정 시점에 dag를 실행합니다.
- Task는 Dag내 어떤 형태로 Workflow가 구성되어 있는 지를 meta DB에 기입을 합니다.
Executor
- 현재 시점에 실행해야하는 Task를 주기적으로 찾습니다.
- 실행해야하는 Task가 있다면, 상태를 변경하고 내부 Queue에 Task를 넘겨주게 됩니다.
Worker
- 전달받은 Task를 실행하게 됩니다.
- 실행하는 Task의 형태는 CLI Command 입니다.
- 다시 말해, CLI Command가 워커에서 실행되는 구조입니다.
Executor의 종류
출처 : https://airflow.apache.org/docs/apache-airflow/stable/executor/index.htmlExecutor는 Task 인스턴스를 실행하는 주체입니다. 종류에 따라, Airflow가 다르게 동작을 합니다.
Airflow Executor에는 Base Executor의 부모 클래스를 상속 받는 여러 Executor가 있습니다.
현업에서 가장 많이 사용되는 것은 Local Executor와 Celery Executor입니다.
LocalExecutor
LocalExecutor는 airflow.cfg 에 있는 parallelism factor의 값의 여부에 따라 두 가지로 나눠지게 됩니다.
- _UnlimitedParallelism
- 해당 값이 0이면 이론 상 Task를 무한히 생성할 수 있습니다.
- worker의 구현체는 Local Worker가 되게 됩니다.
- result_queue 하나만 갖습니다.
- _LimitedParallelism
- 1 이상 경우는 Task는 유한하게 생성할 수 있습니다.
- worker의 구현체는 QueuedLocalWorker가 되게 됩니다.
- task_queue, result_queue 두개를 갖습니다.
고가용성에 문제가 있어서 beta 환경이나 dev 환경에서 사용합니다.
CeleryExecutor
내부적으로 queue를 가지고 있는것은 동일
- Task를 외부 message broker을 통해 비동기로 전송
- 전송된 Task를 Worker Machine의 Celery Worker가 가져가서 CLI Command로 실행
출처 : https://engineering.linecorp.com/ko/blog/data-engineering-with-airflow-k8s-1/KubernetesExecutor
- 다양항 형태의 소스 데이터 환경
- 다양한 형태의 타겟 데이터 환경
- 각각의 서비스에서 만들어지는 데이터의 형태와 환경이 각각이고 다른 경우
- MySQL, MSSQL, NoSQL, byte, 클라우드, 하드 클러스터 등
Airflow on K8s vs Kubernetes Executor
Airflow on Kubernetes
일반적인 Airflow on Kubernetes를 통해 로컬 클러스터를 구성한다면,
- 단순히 웹서버와 스케줄러
- Web server와 Scheduler를 하나의 Pod
- Celery를 포함한 Airflow
- Web server, Scheduler Pod
- Redis or RabbitMq 등 메시지 브로커 cluster를 포함한 Pod
- Airflow Worker Pod
장점
- 구성이 간단
- Template화 하기 쉽다
단점
- 유지보수 비용
- 의존성에 따라 무거워 짐
- Kubernetes의 장점을 살리지 못함
이러한 단점을 극복하기 위해 KubernetesExecuter를 사용
Kubernetes Executer
- 필요할 때만 Kubernetes 자원을 사용
- 필요한 Docker Container만 골라 Pod로 실행
- K8sPodOperator
일반적인 Operator
- 일반적인 BashOperator, PythonOperator은 task가 들어오면 k8sExecutor가 airflow.cfg에 kubernetes session에 관리자가 정의한 k8s config와 이미지 정보를 가지고 worker를 만듭니다.
- Airflow Worker Pod를 생성한 후, 이 Worker Pod가 일반 Operator를 실행하게 되고, 작업이 완료되면 리소스를 반납하고 종료되게 됩니다.
K8sPodOperator
- Airflow Worker Pod를 생성하는 것 까지는 기본적인 Operator와 동일합니다
- Worker에서 개발자가 dag script에 직접 정의한 custom Pod가 수행됩니다
- 다시말해, worker에서 내가 실행하고 싶은 task를 또 다른 Pod로 실행합니다
Kubernetes
일반적으로 쿠버네티스의 경우는 DevOps나 백엔드에서 마이크로서비스 아키텍쳐를 구성할 때 많이 사용합니다.
하지만 데이터 엔지니어링에서도 쿠버네티스는 스케줄러, 태스크의 리소스 관리, 오퍼레이션을 유용하고 쉽게 할 수 있습니다.
Kubernetes + Helm Chart + Github Action + ArgoCD
를 통해 쉽게 k8s CI/CD를 구축하고 관리/운영할 수 있습니다.
