[DevOps] - Apache Airflow

Apache Airflow란?

Basic Airflow deployment

 

  Apache Airflow는 데이터 파이프라인을 자동화하고 관리하는 워크플로우 오케스트레이션 도구이다. 쉽게 말해서, 어떤 작업을 언제, 어떤 순서로 실행할지 결정하고 관리해주는 도구라고 보면 된다.

만약 ETL(Extract, Transform, Load) 작업을 수행해야 하는 데이터 엔지니어라면, Airflow를 사용하면 데이터를 일정 시간마다 자동으로 불러오고 가공하고 저장할 수 있다. 개발자가 수동으로 작업할 필요 없이, Airflow가 지정한 일정에 맞춰 실행할 수 있다.

Apache Airflow의 플러그인 리스트는 여기에서 확인 가능하고, 이외에 자세한 내용은 여기에서 확인할 수 있다.

---

Airflow를 왜 사용할까?

데이터 파이프라인을 운영할 때, 대부분 다음과 같은 문제를 겪거나 고민하게 된다.

• 매번 작업을 수동으로 실행해야 함 → 매일 특정 시간에 데이터를 가져와야 할 때 사람이 직접 실행해야 한다.
• 작업 실행 순서를 관리하기 어려움 → A 작업이 끝난후 B 작업이 실행돼야 하는데, 스크립트로 일일이 관리하기에 번거로움.
• 실패한 작업을 관리하기 어려움 → 특정 작업이 실패하면 자동으로 재시도를 해야 하는데, 기존 cron 스케줄러로는 이런 기능을 지원하지 않는다.

Airflow를 사용하면 위와 같은 문제들을 해결할 수 있다.

• DAG(Directed Acyclic Graph)를 활용해 작업 간의 실행 순서를 명확하게 정의할 수 있음 → ‘작업 실행 순서 관리’ 문제 해결.
• 실패한 작업이 있으면 자동으로 재시도할 수 있고, 실패 내역을 로그에서 쉽게 확인 가능 → ‘실패한 작업 관리’ 및 ‘매번 작업을 수동으로 실행해야 함’ 문제 해결.
• 웹 UI에서 실행 상태를 실시간 모니터링할 수 있다.

---

DAG (Directed Acyclic Graph)

 

병렬 비순환 그래프의 흐름

 

  DAG는 Airflow에서 작업(Task)들의 실행 순서를 정의하는 설계도이다. 위에서 문제점으로 언급된 ‘작업 실행 순서를 관리하기 어려움’ 문제를 해결할 수 있게 도와준다.

DAG는 비순환 그래프 형태로 구성되어 있으며, 특정 작업이 끝나면 다음 작업이 실행될 수 있도록 설정할 수 있다.

 

 예를 들어, “매일 아침 9시에 데이터를 가져오고, 변환한 후 데이터베이스에 저장하는” 파이프라인이 있다고 해보자. 이걸 DAG로 표현하면 다음과 같이 만들 수 있다.

 

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

# 기본 설정 (DAG의 실행 시간 및 재시도 설정)
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2025, 2, 15),  # DAG 시작 날짜
    'retries': 1  # 실패 시 1번 재시도
}

# DAG 정의 (ETL 작업을 자동화할 DAG 생성)
dag = DAG(
    'data_pipeline',  # DAG 이름
    default_args=default_args,
    schedule_interval='@daily'  # 매일 자동 실행
)

# Extract (데이터 추출)
def extract():
    print("데이터 추출: 데이터베이스에서 주문 데이터 가져오기")

# Transform (데이터 변환)
def transform():
    print("데이터 변환: 매출 계산, 날짜 정리, 데이터 클리닝")

# Load (데이터 저장)
def load():
    print("데이터 저장: 변환된 데이터를 데이터 웨어하우스에 저장")

# Task 정의
task1 = PythonOperator(task_id='extract', python_callable=extract, dag=dag)
task2 = PythonOperator(task_id='transform', python_callable=transform, dag=dag)
task3 = PythonOperator(task_id='load', python_callable=load, dag=dag)

# 실행 순서 지정
task1 >> task2 >> task3  # Extract → Transform → Load 순서대로 실행

 

이제 위 DAG가 실행되면 다음과 같은 흐름으로 작업이 진행된다.

1단계 : Extract (데이터 추출)

• Airflow가 extract() 함수를 실행한다
• 데이터베이스에서 주문 데이터를 가져온다

2단계 : Transform (데이터 변환)

• Airflow가 transform() 함수를 실행한다.
• 데이터를 정리하고 매출을 계산하고 날짜별로 정렬한다.

3단계 : Load (데이터 저장)

• Airflow가 load() 함수를 실행한다.
• 정리된 데이터를 데이터 웨어하우스에 저장한다.

이제 이 DAG가 설정된 일정에 따라 자동으로 실행되면서 ETL 파이프라인이 동작하게 된다.

---

Airflow 도입 전 고려사항

 Apache Airflow는 강력한 워크플로우 오케스트레이션 도구지만, 모든 환경에서 최적의 선택은 아닐 수도 있다. 도입 전에 몇몇의 요소를 고려해야 한다.

 

✔ 워크플로우의 복잡성과 필요성

• 현재 운영 중인 배치 작업이 많고 복잡한가?
• 여러 개의 작업을 하나의 파이프라인으로 관리해야 하는가?
• 실패한 작업을 자동으로 재시도하는 기능이 필요한가?

→ 단순한 작업이라면 굳이 Airflow를 도입할 필요가 없고, 복잡한 워크플로우가 필요하다면 Airflow가 강력한 해결책이 될 수 있다!

위의 내용을 포함하여 Airflow 도입 전에 고려해야 할 핵심 체크리스트는 아래와 같다.

고려 요소 체크포인트

워크플로우 복잡성	Airflow가 필요한 수준의 자동화가 필요한가?
인프라 환경	단일 서버 vs 분산 환경 (클라우드, Kubernetes 등)
서버 리소스	CPU, 메모리, 데이터베이스 구성 가능 여부
Executor 선택	LocalExecutor vs CeleryExecutor vs KubernetesExecutor
보안 및 접근 제어	Airflow 웹 UI 접근 제한 및 실행 로그 보안

 

  Airflow는 강력한 데이터 파이프라인 자동화 도구지만, 무작정 도입하면 리소스 낭비나 관리 어려움이 발생할 수 있다. 따라서, 워크플로우 복잡성, 인프라, 실행 환경, 보안 등을 모두 고려한 뒤 도입하는 것이 바람직하다.