Airflow 사용 예
만약 독자가 매일 밤 12시에 트리거시켜야 하는 데이터 파이프라인을 가지고 있다고 가정해보자.
파이프라인을 간략하게 나타내면 아래의 Task 와 비슷하게 구성이 되어있을 것이다.
1. Downloading Data
2. Processing Data
3. 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 는 프로그래밍 방식으로 워크플로우를 작성, 예약 및 모니터링하는 오픈소스 플랫폼입니다.
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 들을 병렬적으로 실행이 가능 (Parallelly executable) 하다. 또한 쿠버네티스 (kubernetes) 클러스터, 분산 클러스터 환경에서도 파이프라이닝이 가능하다.
Useful User Interface
Airflow Webserver 에서 제공해주는 웹 인터페이스를 통해 데이터 파이프라인을 모니터링, 관리하기 매우 편리하다.
High Extensibility
Airflow Instance 에는 손쉽게 Plugin 을 설치할 수 있어서, 새로운 작업 툴이 나와 적용해야 할 필요가 있을 때는 플러그인을 개발하여 적용할 수 있다. 따라서 Airflow 가 특정 툴을 지원할 때까지 기다릴 필요가 없다. 이처럼 Airflow Instance 는 Customizing 하기가 굉장히 쉽다.
Core components
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 ..
Worker : 실제 Task 를 처리하는 컴포넌트
Executor 는 어떻게 실행될지에 대한 정의를 담당, Worker 는 실제 프로세스 작업
Airflow Key Concepts
DAG (Directed Acyclic Graph)
방향을 가진 그래프 중 순환을 포함하지 않은 그래프이다.
Airflow 는 Task 의 연결관계를 DAG 로 관리하고, 웹 인터페이스를 통해서도 DAG 구조를 시각적으로 확인할 수 있다.
따라서 DAG 에 대해 이해하고 있으면 Airflow 를 조금 더 잘 이해하고 사용할 수 있다.
Operator
Task 의 Wrapper 역할
원하는 작업을 달성하기 위해 존재한다
예를 들어, DB에 연결하여 데이터를 Insert 하고 싶다면 Operator 를 이용하여야 작업을 할 수 있다.
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 를 통해서는 오케스트레이션만 진행하는 방식으로 주로 사용한다.
