흔한 유스케이스로 먼저 사용 예를 파악한 후 에어플로우에 대해 자세히 살펴보자.
Airflow 사용 예
만약 독자가 매일 밤 12시에 트리거시켜야하는 데이터 파이프라인을 가지고 있다고 가정해보자.
파이프라인을 간략하게 나타내면 아래의 Task와 비슷하게 구성이 되어 있을 것이다.
- Downloading Data
- Processing Data
- Storing Data
위의 Task들은 실행 순서가 Sequential함이 보장이 되어야한다.
또한 각각의 Task는 서로 다른 외부 Tool과의 interaction이 필요하다.
위의 예에서는
- Data download를 위한 API Request -> 외부 API
- Data process를 위한 Spark Job
- Data store를 위한 Insert / Update 작업 -> 외부 DB
Sequential하게 실행이 되도록 하기 위해서는 외부 API, DB 또한 healthy한 상태에서 Pipeline을 트리거해야 성공적으로 Pipelining을 완료할 수 있다.
만약 Pipelining Task 중에 하나라도 문제가 생기면 어떻게될까?
API가 정상적으로 작동하지 않거나, 메모리 부족으로 Spark에서 문제가 발생하거나, DB에 작업 내용이 저장이 되지 않는다면?
만약 독자가 이러한 데이터 파이프라인을 수십, 수백개를 관리해야한다면?
따라서, Airflow를 이용하면 이러한 문제들을 손쉽게 해결하도록 도와준다.
수많은 데이터 파이프라인을 안정성 높은 방식(reliable한 방식)으로 관리, 실행, 모니터링 할 수 있도록 도와준다.
Airflow의 정의
Apache Airflow is an open source platform to programmatically author, schedule, and monitor workflows
Apache Airflow는 프로그래밍 방식으로 워크플로우를 작성, 예약 및 모니터링하는 오픈 소스 플랫폼입니다.
The Complete Hands-On Introduction to Apache Airflow on Udemy
Airflow is an orchecstrator allowing you to execute your tasks at the right time, in the right way, in the right order.
Airflow는 너의 작업을 정확한 시간에, 정확한 방법으로, 정확한 순서대로 실행하게 해주는 오케스트레이터이다.
Airflow의 장점
🧐 Dynamic Data Pipeline
Airflow는 데이터 파이프라인을 정의하는데, Python을 이용한다.
따라서 Python으로 가능한 대부분의 작업들을, Airflow 파이프라인에서 처리가 가능하다.
🤭 Scalability
Airflow는 매우 뛰어난 확장성을 가지고 있다. 원하는 만큼의 다양한 Task들을 병렬적으로 실행이 가능(Parellelly executable)하다. 또한 쿠버네티스(Kubernetes) 클러스터, 분산 클러스터 환경에서도 파이프라이닝이 가능하다.
😲 Useful User Interface
Airflow Webserver에서 제공해주는 웹 인터페이스를 통해 데이터 파이프라인을 모니터링, 관리하기 매우 편리하다.
😯 High Extensiblility
Airflow Instance에는 손쉽게 Plugin을 설치할 수 있어서, 새로운 작업 툴이 나와 적용해야할 필요가 있을때는 플러그인을 개발하여 적용할 수 있다. 따라서 Airflow가 특정 툴을 지원할 때 까지 기다릴 필요가 없다. 이처럼 Airflow Instance는 Customizing 하기가 굉장히 쉽다.
Core components
이제 Airflow에서 이용되는 코어 컴포넌트들에 대하여 간략히 알아보자.
위 그림은 라인 기술 블로그에서 인용하였습니다.
- Scheduler : Workflow를 스케줄링하는 스케줄러 데몬이다. Airflow에서 가장 핵심이 되는 컴포넌트
- Web Server : 앞서 언급한 Airflow의 웹 인터페이스를 제공하는 웹 서버. Flask와 Gunicorn을 이용하여 인터페이스를 제공
- MetaStore(MetaDB) : 메타데이터가 저장되는 데이터베이스. 주로 Postgresql을 추천하지만, SQL Alchemy와 호환 가능한 MySQL이나 SQLite도 이용이 가능하다.
- Executor : 어떤 환경에서 Task가 실행(Execute)될지에 대한 타입 정의
Debug Executor, Local Executor, Sequential Executor, Celery Executor , etc.. Executor Type Doc - Worker : 실제 Task를 처리하는 컴포넌트
Executor는 어떻게 실행될지에 대한 정의를 담당, Worker는 실제 프로세스 작업
🔑 Airflow Key Concepts
앞으로 Airflow에 대해 학습하며 필요한 필수 개념들을 간략하게 알아보자
DAG (Directed Acyclic Graph)
위 그래프에서 보이는 것 처럼 유향(방향을 가진) 그래프 중 순환을 포함하지 않은 그래프이다.
Airflow는 Task의 연결관계를 DAG로 관리하고, 웹 인터페이스를 통해서도 DAG 구조를 시각적으로 확인할 수 있다.
따라서 DAG에 대해 이해하고 있으면 Airflow를 조금 더 잘 이해하고 사용할 수 있다.
DAG의 자세한 내용은 bossm0n5t3r님의 글을 참고
Operator
Task의 Wrapper 역할
원하는 작업을 달성하기 위해 존재한다
예를 들어, DB에 연결하여 데이터를 Insert 하고 싶다면 Operator를 이용하여야 작업을 할 수 있다.
Operator Type에 대해 알아보자.
Action Operator
기능이나 명령을 실행하는 Operator.
- Bash Operator
- Python Operator
- etc..
Transfer Operator
데이터를 Source에서 Destination으로 전송해주는 Operator.
예를 들어, Presto에서 MySQL로 데이터를 전송하는데에 사용
Sensor Operator
특정 조건을 Sensing하여 실행되는 Operator.
다른 Operator들과는 달리 조건이 만족할 때까지 기다렸다가, 조건이 충족되면 다음 Task를 실행하도록 함.
예를 들어, 특정 위치에 파일이 생성되었을 때 다음 Task를 실행하도록 File sensor를 사용할 수 있음
Task & Task Instance
Task
Task는 데이터 파이프라인에 존재하는 Operator를 의미한다
Task Instance
Task Instance는 데이터 파이프라인이 Trigger되어 실행될 때 생성된 Task를 Task Instance라고 한다.
OOP에 대한 이해가 있다면, Task는 Class, Task Instance는 Object Instance라고 보면 이해가 쉬울 것 같다.
Workflow
워크플로우는 앞서 언급한 모든 개념들을 조합하여 만들어진 개념이다.
DAG를 통해 각 태스크 간의 디펜던시를 정의하고, 각 태스크는 Operator로 실행하는 이러한 일련의 과정들을 통해 워크플로우는 정의된다.
😡 Airflow의 잘못된 용도
Data Streaming Solution
초단위의(또는 더 짧은 주기로) 데이터 처리가 필요한 경우
스트리밍 용도로 에어플로우를 사용하기에는 부적절하다.
Data Processing Framework
에어플로우는 데이터 프로세싱에는 이용하지 않는 것이 좋다. (스파크와는 다르다!!)
그러한 작업에 최적화 되어 있지도 않아서 매우 느리고, 경우에 따라 메모리 부족으로 작업이 진행되지 않을 수도 있다.
따라서, SparkSubmitOperator와 같은 Operator를 이용하여, 데이터 프로세싱은 Spark와 같은 외부 Framework로 처리
Airflow를 통해서는 오케스트레이션만 진행하는 방식으로 주로 사용한다.
다음에는 어떻게 Airflow가 작동하는지 알아보자
