Airflow
Airflow 란, 데이터 파이프라인을 구축하고 유지하는 작업을 쉽게 할 수 있도록 도와주는 python 기반의 오픈 소스 툴.
DAG(Directed Acyclic Graphs)라고 불리는 워크플로우(workflow)를 작성하고, 워크플로우가 특정한 시간에 run 하도록 스케쥴링하고, 이러한 과정이 문제없이 잘 작동하는지 web UI를 통해 모니터링 할 수 있게 해준다.
Airflow Basic Architecture
Metadata Database
dag 정보, 실행이력, 스케줄링 이력, 유저 정보 등 실행하는 workflow에 필요한 메타데이터를 저장한다.
Airflow Scheduler
DAG 분석, 현재 시점에서 DAG의 스케쥴이 지난경우 Airflow 워커에 DAG의 태스크를 예약.
Airflow Worker
예약된 태스크를 선택하고 실행
Airflow Web Server
스케쥴러에서 분석한 DAG를 시각화하고 DAG 실행과 결과를 확인할 수 있는 주요 인터페이스 제공
Airflow Executor
위 그림에서는 executor와 scheduler가 분리되어있지만, 사실 executor는 scheduler 안에서 동작한다. 작업을 worker가 수행하는 방식을 결정한다.
DAG
python code로 만들어진 data pipeline 정의 파일로, 이것 역시 웹서버, 스케줄러, 워커가 접근할 수 있어야 한다.
airflow.cfg
configuration file로 timezone, database connection 등 실행에 필요한 다양한 정보를 저장한다. 웹서버, 스케줄러, 워커가 접근할 수 있어야 한다.
DAG
Directed Acyclic Graph. 방향성 비순환 그래프
실행하고 싶은 Task들의 관계와 dependency를 표현하고 있는 Task들의 모음. 어떤 순서와 어떤 dependency로 실행할지, 어떤 스케줄로 실행할지 등의 정보를 가지고 있음.
Airflow Operator
각 Airflow DAG는 여러 Task로 이루어져있습니다. operator나 sensor가 하나의 Task로 만들어집니다. Airflow는 기본적인 Task를 위해 다양한 operator를 제공합니다.
- BashOperator : bash command를 실행
- PythonOperator : Python 함수를 실행
- EmailOperator : Email을 발송
MySqlOperator : sql 쿼리를 수행
Airflow 2부터는 핵심 오퍼레이터만 남음
Airflow Executor
DAG 프로세서 및 태스크 스케줄러는 Airflow 스케줄러 시작할 때 한 가지 방법에 의해서만 실행 가능. 설치 시점에 환경이 고정됨.
현재 Airflow Executor에는 네가지 존재.
- SequentialExecutor(기본값)
- LocalExecutor
- CeleryExecutor
- KubernetesExecutor
SequentialExecutor
Airflow에서 제공하는 기본 Executor로 sqlite와 함께 사용할 수 있는 Executor로 한번에 하나의 task만 실행할 수 있어 병렬성을 제공하지 않아 실제 운영환경에는 적합하지 않음.
데이터베이스로 sqlite는 동시 쓰기가 불가능하여 운영환경에 적합하지 않음. 따라서 SequentialExecutor에서만 사용되고, 다른 모든 Executor에서는 MySQL 이나 PostgresSQL을 사용
LocalExecutor
Sequential Executor과 달리 task를 병렬로 실행하는 것이 가능하며, 옵션값을 통해 최대 몇 개의 task를 병렬로 실행할지 설정하는 것이 가능하다
AIRFLOW__CORE__PARALLELISM, AIRFLOW__CORE__DAG_CONCURRENCY, AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG 값으로 병렬 태스크 수 제한 가능.
해당 환경은 스케줄러 시스템의 리소스에 의해 제한되며, 이때 LocalExecutor가 더 이상 프로세스를 늘릴 수 없는 상황이 발생하면 CeleryExecutor나 KubernetesExecutor 고려해볼 수 있음.
CeleryExecutor
여러 개의 워커(서로 다른 서버)로 클러스터를 구성하고 대기열(Broker) 시스템 (RabbitMQ나 Redis 등)을 통해 워커에게 메시지를 전달하여 task를 각 워커에서 실행.
HA(고가용성)과 Scale Out이 가능한 장점.
DAG 폴더에 웹서버, 모든 워커, 스케쥴러가 접근 가능하도록 공유 파일 시스템이나 git-sync 등의 전략을 사용해야함.
Flower라는 익스큐터, 태스크 및 전체 Celery 시스템 상태를 검사할 수 있는 웹 기반 모니터링 도구 추가 제공.
KubernetesExecutor
Airflow 워커를 pod 형태로 실행. 매 Task 마다 pod가 생성되어 가볍고, Worker에 대한 유지 보수 필요없다는 장점.
K8S를 활용하여 지속적으로 자원을 점유하지 않아 효율적으로 자원 사용 가능.
이미지 버전관리나 pod 재배포가 쉬움.
모든 worker pod에서 DAG 코드를 가지고 있어야 하기 때문에, git-sync나 shared volume 등을 활용해야함.
Worker pod가 종료되면 Task에 대한 로그파일도 삭제됨. 외부저장소(S3 등)을 이용하여 로깅해야함.
pod를 생성하는 overhead가 존재하고, 구성이 복잡함.
Airflow vs jenkins
Airflow는 데이터 ELT(추출, 변환, 적재)을 위한 워크플로우를 만들고 관리하는 목적.
Jenkins는 CI 툴로, 워크플로우를 작성하고 관리하는 것만을 목적으로 만들어진 것이 아님. ELT 구성을 위해서는 필요에 따라 다양한 플러그인 연동 필요.
또한, Airflow에서는 다양한 분산 환경에서의 데이터 파이프라인 구축이 가능하다. (다양한 외부 리소스 연동 가능)
💡 복잡한 데이터 파이프라인 관리가 필요하다면 Airflow 복잡한 데이터 파이프라인이 아닌 단순 스케쥴 처리 작업이면 jenkins를 쓰자
Airflow vs 스프링배치
| 구분 | 스프링 배치 | Airflow |
| 워크플로 | Job | DAG |
| 단일 작업 | Tasklet 기반 작업 | 단일 Operator or Task |
| 단순 데이터 파이프라인 | 아이템 Chunk 기반의 Step ItemReader - ItemProcessor -ItemWriter |
복수개의 Operator 또는 Task를 활용 아이템 리더 태스크 >> 아이템 처리 태스크 >> 아이템 쓰기 태스크 |
| 복잡한 데이터 파이프라인 | 복수 개의 Steps | 복수개의 Operator 또는 Task를 활용 각 Operator 또는 Task 의존 연결 |
스프링 배치와 Airflow 둘다 파이프라인 구성이 가능하다.
주로, 도메인 단위 배치 업무는 스프링 배치에서, 크로스 도메인 단위 배치 업무는 Airflow에서 처리하는 걸 추천한다.
💡 스프링 배치 파이프라인은 GUI 툴로 확인이 어려우니, 복잡한 파이프라인 구성의 경우 Airflow에서 처리하는게 GUI 툴로 확인 가능해서 좋을 것 같음.
참고
http://www.yes24.com/Product/Goods/107878326
https://www.bucketplace.com/post/2021-04-13-버킷플레이스-airflow-도입기/
https://velog.io/@hamdoe/Airflow-조그맣게-시작하기-소개
https://velog.io/@jaylnne/ETL-Airflow-와-Jenkins-비교-무엇을-사용할까
'DEV > Airflow' 카테고리의 다른 글
| 기타 필수 개념 (0) | 2023.04.17 |
|---|---|
| DAG 예시 및 DAG 반영 참고 (0) | 2023.04.17 |
댓글