📝서론
3월이 되면서 담당하는 서비스를 대폭(?) 늘리게 되었는데, Managed Workflows for Apahce Airflow도 그 대상 중 하나였습니다. 다른 서비스들에 비해 AWS에서 출시된지 그리 오래되지 않았음에도 불구하고 최근 들어 꽤나 많이 쓰이는 것으로 느껴졌는데요. 잘 정리해두면 도움이 될 것 같아 이렇게 정리 글을 작성하게 되었습니다.
🤜본론
Apache Airflow란?
아파치 에어플로(Apache Airflow)는 데이터 공학 파이프라인을 위한 오픈 소스 워크플로 관리 플랫폼이다. 2014년 10월 에어비앤비에서 기업의 점차 복잡해지는 워크플로를 관리하기 위한 해결책으로서 시작하였다.
(출처 위키백과)
말 그대로 워크플로우를 관리해주는 일종의 툴이라고 생각하면 편할 것 같습니다. ETL 작업을 생각해보면 추출,변환,로드의 흐름이 있는데 이 수 많은 workflow를 편리하게 관리할 수 있게 해주는 툴이 바로 Airflow 입니다.
Airflow의 장점
Workflow를 하나의 프로그램 (Python)으로 작성할 수 있는 플랫폼입니다. 파이썬을 이용하기 때문에, 파이썬을 통해 작업할 수 있는 작업들을 대부분 처리할 수 있으며 파이썬의 많은 라이브러리를 이용할 수 있다는 장점도 있습니다.
또한 Web UI를 제공하기 때문에 손 쉽게 모니터링을 할 수 있습니다.
확장성 역시 Airflow의 장점인데요, 플러그인 설치를 쉽게 할 수 있기 때문에 필요한대로 커스텀하기에도 좋습니다.
Managed Workflows for Apache Airflow
AWS에서 제공하는 Managed Workflows for Apahce Airflow(이하 MWAA)는 관리형 서비스이기 때문에 직접 Airflow를 설치하고 유지 및 관리하는데에 쓰이는 리소스를 줄이게끔 도와줍니다. 또한 각종 AWS 서비스와 통합이 아주 편리합니다.
🌪️ Airflow 주요 components
DAG(Directed Acyclic Graph)
순환이 없고, 순서가 정해진 방향 그래프를 의미합니다. 상대적인 순서가 있기 때문에, 예를 들면 1이 2와 3보다는 먼저 수행이 됩니다.
Airflow web UI를 통해서 다음과 같이 그래프를 시각적으로도 확인할 수 있습니다. 각 색깔로 Task가 수행이 완료 되었는지, Running 상태인지도 확인할 수 있습니다.
DAG Runs
DAG가 실행 될 적에, DAG Runs가 다음과 같이 생성되어 관리됩니다.
DAG Runs의 상태는 아래와 같이 결정됩니다.
- 예정된 스케줄에 의하여 DAG가 실행될 조건이 만족될 경우, Queued 상태의 DAG run 생성, scheduler는 DAG Run의 상태를 Running으로 전환시킴.
- DAG의 최종 상태는 DAG 내의 Task에 의해 결정
- DAG 내의 하나의 Task라도 Failed 되면 DAG Run의 상태도 Failed
- DAG 내의 모든 Task가 success이면 DAG Run의 상태도 Success
Task
순서 때문에 위에 Task라는 용어가 먼저 등장했지만, 간략하게 이야기하면 Task는 Aiflow에서의 execution의 기본 단위입니다.
Operator
사전에 정의된 Task의 템플릿이고, 작업을 하기 위해 operator를 사용합니다. 종류로는 작업을 수행하거나 다른 시스템에서 작업이 수행되게끔 하는 Action Operator와 특정 시스템에서 다른 시스템으로 데이터를 옮기는 Transfer Operator가 있습니다.
...(omitted)...
from airflow.providers.amazon.aws.operators.glue_crawler import AwsGlueCrawlerOperator
...(omitted)...
glue_crawler = AwsGlueCrawlerOperator(
task_id="glue_crawler",
config=config,
dag=dag)이렇게 AWS Glue의 크롤러를 업데이트 또는 생성을 트리거하게끔 할 수 있습니다. 더 많은 operator들은 아래를 참고해주세요.
참고 : https://airflow.apache.org/docs/apache-airflow/2.2.3/_api/airflow/operators/index.html
Task들로 DAG 구성하기
비트 연산자(>>, <<)를 이용해 DAG를 표현할 수 있습니다.
예시는 아래와 같습니다.
task1 수행 후 task2가 수행되게끔 할 때
task1 >> task2
또는 set_upstream(), set_downstream() 함수를 이용할 수 있다.
task1.set_downstream(task2)
Task Instance
Task가 실행될 때 생성되어 관리됩니다. 객체 지향 개념으로 따지면 Task instance는 Task가 인스턴스화 되어 실행되는 것이라고 생각하면 편리할 것 같습니다.
Scheduler
Scheduler는 이름만 들어도 어떤 역할을 올지 감이 잡힙니다 ㅎㅎ Airflow에서 가장 핵심적인 역할을 하고 있습니다. Workflow를 스케줄링하는 daemon입니다.
heartbeat
Scheduler의 heartbeat 메서드는 Scheduler가 정상적으로 작동하는지 확인하며, 자신의 상태를 DB에 업로드합니다.
Executor
Task를 실행시키는 역할을 합니다. 크게 Local Executor와 Remote Executor가 있습니다. Executor에 대한 자세한 내용은 아래 문서를 확인해주세요.
참고 : https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/executor/index.html
우리는 Executor 타입 중 Remote executor에 속하는 Celery Executor를 알아볼 필요가 있습니다. 그이유는 MWAA가 Celery Executor를 이용하기 때문입니다.
Celery Executor(Celery Worker)
이미지 출처 : 오늘의 집 블로그
<구성요소>
- Result Backend (완료된 command 상태를 저장)
- Broker (실행 될 command를 저장)
MWAA에서는..
작업을 병렬로 실행하고 각 worker들이 동시에 실행될 수 있도록 환경을 셋팅합니다. 작업이 Queue에 추가되면 MWAA에서는 환경 생성시 결정한 최대 worker의 수에 도달할 때 까지 worker를 추가합니다.
만약 10개의 worker가 max worker수로 설정되어있다면, 더 실행할 Task의 수가 없을 때 까지 추가 executor들을 계속해서 실행하고, 더 실행할 Task가 없으면 worker들을 폐기하는 방식으로 Auto Scaling을 합니다.
Database
메타 데이터들을 저장하는 데이터베이스입니다. 주로 PostgreSQL을 이용하나 MySQL이나 SQLite도 사용할 수 있습니다.
MWAA에서는..
환경을 생성할 때 2개의 private subnet을 선택하게 되는데요, 각각의 subnet에 AWS Managed PostgreSQL DB와 Fargate 컨테이너를 생성합니다. 만약 하나의 zone에서 장애가 생기면 다른 zone에 있는 컨테이너로 전환 후 Celery 실행기를 실행 후 스케줄러가 PostgreSQL DB에 새 task instance를 생성합니다. 이 DB에는 직접적으로는 접근이 불가하지만, DAG를 이용하여 DB cleanup 작업을 수행할 수 있습니다. 자세한건 여기를 참고해주세요.
Web Server
Airflow에서 제공되는 Web UI를 통해 다양한 모드의 view를 통해 모니터링이 가능합니다.(이미지 출처 : apache airflow 공식)
DAGs View
Grid View
Graph View
Tree View
Airflow에서 중요한 컴포넌트들에 대해서 정리해보았습니다. 내용이 너무 길어져서, 다음 2탄에서는 간단하게 MWAA를 통해 DAG를 실행해보는 예제를 준비하려고 합니다. 읽어주셔서 감사합니다. (ㅡ ㅡ)
