[Airflow] Apache Airflow란?

 

 

이번 글에서는 `Apache Airflow`에 대해 정리하려고 한다.

 

최근 데이터 파이프라인의 복잡성이 증가함에 따라, 이를 효과적으로 관리하고 자동화할 수 있는 도구에 대한 수요가 높아지고 있다.

 

그 중 `Apache Airflow`가 있으며, 이는 다양한 곳에서 널리 사용되고 있다.

 

그렇다면 `Apache Airflow`는 무엇일까?

 

What is Apache Airflow?

---

먼저, `Apache Airflow`에 대해 공식 홈페이지는 아래와 같이 정의하고 있다.

 

an open-source platform for developing, scheduling, and monitoring batch-oriented workflows

 

 

즉, `Airflow`는 배치 중심 워크플로우를 개발, 스케줄링 및 모니터링할 수 있는 오픈소스 플랫폼이다.

 

이는 코드 기반의 워크플로우 정의, 스케줄링 기능, 다양한 외부 시스템과의 연동 등을 통해 데이터 처리 및 `ETL`(Extract, Transform, Load) 프로세스를 효율적으로 운영할 수 있다.

 

또한, `Python` 기반으로 작성되며, 다양한 외부 시스템과의 연동을 지원하여 유연하게 확장할 수 있다.

 

---

 

`Airflow`는 다음과 같은 장점이 있다.

1. 코드 기반 워크플로우 정의: `Python`을 사용하여 워크플로우를 정의할 수 있어 유지보수와 확장이 용이
2. 스케줄링 및 모니터링: 다양한 실행 일정과 실패 처리 로직 설정 가능
3. 유연한 확장성: `CeleryExecutor`, `KubernetesExecutor` 등을 활용하여 수평 확장 가능
4. 시각화 도구: 웹 UI를 통해 실행 상태를 직관적으로 확인 가능
5. 다양한 외부 시스템 연동: 다양한 데이터 소스 및 도구와 통합

 

What is ETL(Extract, Transform, Load)?

---

ETL.

 

앞서 `ETL`에 대해 언급했었는데 이에 대해 알아보자

 

`ETL`(Extract, Transform, Load)은 데이터를 수집, 변환, 적재하는 과정으로, 데이터 분석과 머신러닝을 비롯한 다양한 데이터 활용 분야에서 필수적인 절차이다.

• `Extract(추출)`: 원본 데이터 소스에서 데이터를 수집하는 단계
• `Transform(변환)`: 데이터를 정제하고 가공하여 원하는 형식으로 변환하는 단계
• `Load(적재)`: 변환된 데이터를 데이터 웨어하우스, 데이터베이스 또는 분석 시스템에 저장하는 단계

즉, 데이터 활용 과정에서 원하는 정보를 얻기 위해서는 ETL 작업이 필수적으로 수행되어야 한다.

 

`Airflow`는 이러한 `ETL` 프로세스를 자동화하고, 복잡한 데이터 파이프라인을 효과적으로 관리하는 데 최적화된 도구이다.

 

그렇다고 `Airflow`가 `ETL` tool인 것은 아니다. `Airflow`의 목적은 사용자가 정의한 순서와 흐름에 따라 데이터를 처리하는 작업을 처리하는 것이다.

 

 

Apache Airflow Components

---

 

이제 `Airflow`의 주요 구성요소에 대해 알아보자

 

일반적으로 다음과 같은 5가지의 주요 구성 요소로 구성된다.

• `Web Server`
• `Scheduler`
• `Executor`
• `Workers`
• `Metastore`

각각에 대해 알아보자

Web server

`Airflow`는 웹 기반 UI를 제공하며, `Web Server`는 사용자가 관리하고 모니터링할 수 있도록 돕는 역할을 한다.

Scheduler

`Scheduler`는 정의된 작업을 실행 스케줄에 맞춰 실행하는 핵심 구성 요소이다. 정의된 작업에서 실행할 작업의 우선 순위를 결정하고, `Executor`에게 실행할 작업을 전달한다.

Executor

`Executor`는 `Scheduler`가 전달한 작업을 실제로 실행하는 역할을 한다. LocalExecutor, CeleryExecutor, KubernetesExecutor 등 다양한 실행 방식을 지원하며, 환경에 맞게 설정할 수 있다.

Workers

`Worker`는 `Executor`의 지시에 따라 개별 작업을 실행하는 노드들이다. 분산 환경에서 여러 개의 `Worker`를 활용하면 대규모 병렬 처리가 가능하다.

Metastore

`Metastore`는 `Airflow` 메타데이터를 저장하는 데이터베이스이다. 작업 실행 이력, 작업 상태, 사용자 정보등을 저장하며, 보통 `PostgreSQL`이나 `MySQL`을 사용하여 관리한다.

 

Apache Airflow 개념

---

Airflow를 사용하기 위해서는 `DAG(Directed Acyclic Graph)`, `Task`, `Operator`등을 알아야한다.

 

각각에 대해 알아보자

 

DAG(Directed Acyclic Graph)

DAG(Directed Acyclic Graph).

 

`Airflow`는 정의된 작업을 `DAG(Directed Acyclic Graph)` 형태로 관리한다.

 

`DAG`란 위 그림처럼 방향을 가진 그래프 중 순환을 포함하지 않은 그래프이다. 즉, 한 노드에서 출발하여 다시 해당 노드로 돌아오는 경로가 존재하지 않는다.

 

`DAG`의 노드를 `Task`라고 생각하면, `DAG`의 특성 덕분에 데이터 파이프라인을 구성할 때 각 `Task`가 순차적으로 실행되도록 보장할 수 있다.

 

만약 추출하는 Task, 변환하는 Task, 적재하는 Task가 있다고 했을 때 다음과 같은 `DAG` 형태의 작업이 생성되게 된다.

 

 

 

---

 

정리해보면 `Airflow`는 작업을 `DAG` 형태로 관리하며, 이러한 `DAG`는 여러 개의 `Task`로 이루어져있고, `Task` 간의 의존성을 설정할 수 있다.

 

또한, `Airflow`에서는 `DAG`를 코드로 정의하기 때문에 복잡한 워크플로우도 유연하게 구성할 수 있다.

 

`DAG` 파일은 `Python` 스크립트로 작성되며, 보통 다음과 같은 요소를 포함한다.

• DAG 객체: DAG의 이름, 스케줄 주기, 기본 인자 등을 정의
• Task: Operator를 사용하여 개별 작업을 정의
• Task Dependency: `>>` , `<<` 연산자를 사용하여 Task간의 실행 순서를 설정

 

Task

Task.

 

`Task`는 `DAG`내에서 실행되는 개별 작업을 의미한다. `DAG`는 여러 개의 `Task`로 이루어져있으며, 각 `Task`는 특정한 작업 단위를 수행한다.

 

이러한 `Task`는 `Operator`를 통해 정의 되며, `Operator`의 종류에 따라 실행되는 작업의 형태가 달라진다.

 

 

Operator

`Operator`는 `Task`의 실행 방식을 정의하는 역할을 한다. `Airflow`에서 제공하는 대표적인 `Operator`는 다음과 같다.

• `BashOperator`: Bash 명령어를 실행하는 Task를 정의할 때 사용
• `PythonOperator`: Python 함수를 실행하는 Task를 정의할 때 사용
• `SqlOperator`: 데이터베이스에 SQL 쿼리를 실행하는 Task를 정의할 때 사용
• `DummyOperator`: 단순히 파이프라인의 흐름을 조정하는 용도로 사용
• `EmailOperator`: 이메일을 보낼 때 사용
• `Sensor`: 특정 조건이 충족될 때까지 대기하는 Task를 정의할 때 사용

 

DAG 작성법

---

앞서 설명했듯이, DAG는 Python 스크립트로 작성되며 아래와 같이 작성할 수 있다.

 

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def print_hello():
    print("Hello, Airflow!")

def print_goodbye():
    print("Goodbye, Airflow!")

with DAG(
    dag_id='example_python_operator',
    schedule_interval='@daily',
    start_date=datetime(2025, 1, 1),
    catchup=False
) as dag:
    # 첫 번째 Task: "Hello, Airflow!" 출력
    task1 = PythonOperator(
        task_id='print_hello_task',
        python_callable=print_hello
    )
    
    # 두 번째 Task: "Goodbye, Airflow!" 출력
    task2 = PythonOperator(
        task_id='print_goodbye_task',
        python_callable=print_goodbye
    )

    # 작업 연결
    task1 >> task2

 

이 코드는 `PythonOperator`를 사용하여 두 개의 간단한 `Task`를 연결해 `DAG`를 정의하는 예시이다.

 

코드에 자세한 설명에 대해서는 추후에 Airflow 실습 글을 작성할 때 다룰 예정이다.

 

마무리

---

이번 글에서는 Apache Airflow에 대해 정리해보았다.

 

다음 글로는 이러한 Airflow 설치와 실습에 대해 포스팅할 예정이다.