Airflow
Airflow는 Python기반의 workflow scheduler이다. 오픈 소스 플랫폼으로 배치 작업을 개발, 스케줄링, 모니터링할 수 있다.
DAG(Directed Acyclic Graph)은 자료구조에서 본 그 순환하지 않는 방향이 존재하는 그래프(DAG)가 맞다.
Airflow에서 DAG은 하나의 워크플로우 파이프라인이며, DAG이라는 특성상, 반복이나 순환을 허용하지 않는다.(비순환성)
- 노드(Task): DAG의 각 노드(여기서는 A~G)는 Task로, 실행할 작업을 의미한다. Task는 Python 함수, Bash 스크립트, SQL 쿼리 등 다양한 형태로 정의될 수 있다.
- 간선(Edge): DAG에서 간선은 Task 간의 의존성을 나타낸다. 즉, 어떤 Task가 먼저 실행되어야 하고, 그 결과에 따른 다른 Task가 이어서 실행될지를 정의한다. 이를 통해 Task 간의 실행 순서를 명확하게 규정할 수 있다.
Airflow는 어떤 장점이 있길래 현재 스케줄링, 워크플로우 도구의 표준이 될 수 있었을까? Airflow의 단점은 뭘까?🤔
Airflow의 장점
1. 유연한 작업 스케줄링
- Airflow는 복잡한 Batch 작업 스케줄링 및 종속성을 정의하기가 쉽다. DAG(Directed Acyclic Graph)을 사용해 작업 간의 관계를 쉽게 설정할 수 있고, 작업이 언제 실행돼야 하는지 정확하게 제어할 수 있다.
2. 모니터링 및 로깅
- Airflow는 작업 실행 로그 및 매트릭을 모니터링하고, Web UI(대시보드)를 통해 다양한 배치작업을 쉽게 시각화하거나 디버깅할 수 있다. 즉, DAG의 상태를 실시간으로 확인하며, 작업(Task)의 성공, 실패, 재시도 여부를 쉽게 파악할 수 있다.
3. 동시성
- 여러 사람이 동시에 workflow를 개발하고, 개별 workflow별로 동작과 설정 등을 관리할 수 있다.
4. 버전 관리
- Python 코드로 workflow를 작성하기 때문에 VCS(Version Control System)를 이용해 버전 관리를 할 수 있어, 변경 사항을 추적하고 롤백하기 쉽다.
5. 테스트
- 간단한 Unit Test부터 복잡한 Integration Test까지 빠르고 쉽게 할 수 있다.
이외에도 Airflow 등장 이전의 cron이 갖는 문제점(작업 실패 시 재실행x, 알림 기능 제공 x, 실행 로그 확인 불편함)을 해결한다.
Airflow의 단점
1. streaming 작업 불가능
- Airflow는 Streaming 솔루션이 아니다. Airflow는 주로 정해진 스케줄에 따라 배치 작업을 실행하도록 설계됐다. 하지만, Apache Kafka 같은 Streaming 시스템과 종종 함께 사용되어 스트리밍 데이터를 배치 작업으로 처리하는데 활용될 수도 있다.
따라서, 만약 스트리밍 데이터 처리가 주요한 요구 사항이라면, Airflow보다는 Apache Kafka, Apache Flink, Apache Storm 같은 스트리밍 프레임워크를 사용하는 것이 더 적절하다.
2. 무한히 실행되는 작업
- Airflow는 무한히 실행되는 event-based workflow를 위해 설계되지 않았다. 예를 들어, 24시간 내내내 실시간으로 데이터를 처리하는 작업이나, 이벤트가 발생할 때마다 즉시 반응하는 작업과 같이 계속해서 끊임없이 실행되는 작업을 수행하기에 적합하지 않다.
3. Airflow 외부 요소에 의해 trigger되는 scheduling 방식은 사용할 수 없다.
- Airflow는 기본적으로 시간 기반 스케줄링에 최적화돼있으며, 외부 이벤트에 의해 트리거되는 작업은 제한적이다.
따라서, 다음과 같은 작업에 Airflow은 적절하지 않다.
1. 지연을 허용하지 않는 작업의 스케줄링
- Airflow는 특정 시간에 작업이 시작되지만, 그 작업이 정확히 해당 시간에 완료된다는 보장을 제공하지 않는다.
이는 Airflow 아키텍처와 Python 기반의 워크플로우 관리 시스템 특성으로 인해, 작업이 로딩되거나 실행되는 과정에서 지연이 발생할 수 있고, 작업 실패의 가능성도 항상 존재한다.
2. Airflow worker 내에서 고부하 작업 처리
- worker내에서 매우 많은 연산, aggregation 등은 권장되지 않는다. Airflow는 워크플로우 관리에 중점을 두고 있기 때문에, 대규모 데이터를 처리하거나 고부하 작업을 수행할 때는 Apache Spark와 같은 분산 처리 시스템과 통합하여 사용하는 것이 효율적이다.
Airflow의 아키텍처
1. Scheduler
- 스케줄러는 Airflow에서 작업(Task)를 실행할 시점을 관리한다. 각 작업이 언제 실행돼야 하는지 결정하고, 이를 실행할 Executor에게 전달한다. 주기적으로(default, 1분마다) DAG Directory에 있는 모든 DAG(워크플로우) 파일을 확인한다. 각 DAG 파일을 검사해, 특정 작업이 실행될 시간이 됐는지 판단한다. 만약 작업이 실행될 시간이 됐다고 판단되면, 스케줄러는 해당 작업을 실행하기 위해 Executor에게 작업을 전달하고, Executor가 이 작업을 실제로 수행할 수 있도록 트리거한다.
또한, 스케줄러는 DAG 정의와 각 Task의 상태를 Metadata Database에 저장한다. 이후, DAG, Task 상태 관리, 스케줄링 등에 Metadata Database를 활용해 DAG과 작업(Task)의 상태를 추적하고, 작업 스케줄을 관리한다.
2. Executor
- Executor는 Task의 실행을 관리한다. 기본적으로 Airflow 설치 시 모든 Task는 Scheduler가 관리하는 하나의 환경에서 실행된다.
하지만, 실제 운영환경에서는 Executor가 작업을 수행하는 외부의 Worker에게 작업을 할당하여 실행을 분산시킬 수 있다.
(Executor는 Task를 직접 실행할 수도 있고, Worker에게 할당할 수도 있다.)
<Executor의 종류>
- Local Executor: 로컬 환경에서 여러 개의 process를 실행시켜서 Task를 병렬로 처리(Worker가 별도로 존재하지 않아, Executor가 직접 처리). 로컬 머신의 리소스에 의존하기 때문에 작업량이 많아지면 병목이 발생할 수 있고, 확장성이 부족해 대규모 분산처리가 필요한 환경에 적합하지 않음.(주로 개발 환경, 소규모 테스트 또는 초기 배포 환경에서 사용)
- Sequential Executor: Local Executor와 비슷하지만, 로컬 환경에서 한 번에 하나의 Task만 순차적으로 실행(default)
(동시에 여러 Task를 실행할 수 없기 때문에, 대기 시간이 길어지므로 매우 간단한 테스트 환경에서 사용) - Celery Executor: Celery를 이용해서 여러 Worker에게 작업을 분배하고, 작업 결과를 관리 및 분산 처리 가능
- CeleryKubernetes Executor: K8S 환경에서 Celery를 사용여 작업을 분산 처리
- Dask Executor: Dask(python을 이용한 분산 작업 관리시스템)을 이용해 Task의 실행과 결과를 관리
- Kubernetes Executor: K8S환경에서, Task를 Pod(=Worker)에 스케줄링하는 방식으로 동작. 별도의 Pod이 각 Task를 실행
- LocalKubernetes Executor: Task를 로컬 환경에서는 LocalExecutor로, K8S 환경에서는 KubernetesExecutor로 동작시키는 혼합 방식
더 자세한 내용은 공식문서에서 확인할 수 있고, 실제 프로젝트할 때는 지금 내 환경과 수행하는 작업에 맞춰 사용하면 된다.
Local Executor와 Sequential Executor는 실무에서 거의 사용되지 않고,
여러 대의 서버를 직접 관리하면서 작업을 나누어 처리해야 할 때(직접 Machine Pooling)는 Celery Executor,
k8s에서 작업을 컨테이너(Pod)로 분산 처리할 때는 Kubernetes Executor를 사용한다.
3. Webserver
- Airflow의 Web UI를 제공하는 서버로, DAG과 Task의 상태를 시각적으로 확인하고(모니터링), trigger하며 디버깅하는데 유용하다.
4. DAG Directory
- DAG Directory에는 Airflow에서 Task 및 Workflow를 정의하는 DAG 파일들이 저장되는 디렉토리이다. Scheduler와 Executor를 통해 DAG 파일들을 읽고, 실행될 Task를 정의, 스케줄링, 실행한다. 따라서, DAG 파일은 Sceduler와 Executor가 읽을 수 있는 dags 경로에 있어야 한다.
5. Metadata Database
- Metadata Database는 Airflow에서 DAG와 Task의 정의, 상태, 실행 정보, 결과, 로그 등을 저장하는 핵심 데이터베이스로, Scheduler, Executor, Webserver가 이 데이터를 활용해 작업을 관리하고 모니터링한다.
<사용 가능한 Database 종류와 버전>
- PostgreSQL: 11, 12, 13, 14, 15
- MySQL: 5.7, 8
- MSSQL: 2017, 2019
- SQLite: 3.15.0+
Airflow 동작 원리
개발자는 DAG Directory에 DAG 파일을 저장한다. 그럼 Airflow는 개발자가 작성한 Python으로 작성된 DAG 파일을 읽고, Scheduler는 거기에 맞춰 Task를 스케줄링해 Executor에 전달한다. 최종적으로 Executor에 의해 Task가 Worker에게 할당된다.(Worker는 확장 가능해서 동시에 수많은 Workflow를 실행하고 관리할 수 있다.) Task의 실행상태는 Metadata Database에 저장되고, 사용자는 Webserver(UI)를 통해서 각 Task의 실행 상태, 성공 여부 등을 모니터링 및 관리할 수 있다.
Reference
https://airflow.apache.org/docs/apache-airflow/stable/index.html
https://www.bucketplace.com/post/2021-04-13-버킷플레이스-airflow-도입기/
https://fastcampus.co.kr/data_online_engineering
'Data Engineering > Airflow' 카테고리의 다른 글
| Architecture on Celery Executor (0) | 2024.08.19 |
|---|