[Apache Airflow] 데이터 파이프라인을 자동화하는 강력한 도구 !!

 
 

 

GA + Airflow 사용 ==> 집계 통계로 사용

 

1. Apache Airflow란?

Apache Airflow는 워크플로우(Workflow) 및 데이터 파이프라인(Data Pipeline)의 작성, 스케줄링, 모니터링을 자동화하는 오픈 소스 플랫폼입니다. Python 기반으로 개발되었으며, DAG(Directed Acyclic Graph)를 이용해 작업의 흐름을 정의할 수 있습니다.
2014년 Airbnb에서 내부 도구로 개발된 후, 2016년에 Apache Software Foundation의 인큐베이션 프로젝트로 편입되었고, 현재는 정식 Apache 프로젝트로 활발하게 유지보수되고 있습니다.
 

2. Apache Airflow의 주요 특징

2.1 DAG(Directed Acyclic Graph) 기반 워크플로우

Airflow에서 작업 흐름은 DAG(방향성 비순환 그래프)로 정의됩니다. DAG는 개별 작업(Task) 간의 종속성을 나타내며, 각 작업은 특정 실행 순서를 따릅니다.

2.2 다양한 오퍼레이터(Operators)

Airflow는 다양한 내장 오퍼레이터를 제공하여, 간단한 Python 코드 실행부터 데이터베이스 쿼리, 클라우드 서비스와의 연동까지 지원합니다.

• PythonOperator: Python 함수를 실행
• BashOperator: Bash 스크립트 실행
• MySqlOperator, PostgresOperator: SQL 쿼리 실행
• Sensor: 특정 이벤트 발생을 감지하여 작업 실행
• KubernetesPodOperator: 쿠버네티스 환경에서 작업 실행

2.3 강력한 스케줄링 기능

Airflow는 크론(Cron) 스타일의 스케줄링을 지원하며, 작업을 시간 기반 또는 이벤트 기반으로 실행할 수 있습니다.

2.4 모니터링 및 로깅

Airflow는 웹 UI를 제공하여, DAG의 실행 상태를 시각적으로 확인하고, 로그를 통해 문제를 진단할 수 있습니다.

2.5 확장성 및 유연성

Airflow는 플러그인 시스템을 지원하며, 다양한 클라우드 및 온프레미스 환경과 쉽게 통합될 수 있습니다.

 

3. Apache Airflow 아키텍처

Airflow는 여러 개의 주요 컴포넌트로 구성됩니다.

3.1 Web Server

사용자가 DAG을 모니터링하고, 실행 상태를 확인할 수 있도록 하는 웹 기반 인터페이스입니다.

3.2 Scheduler

DAG을 주기적으로 실행하고, 종속성을 고려하여 적절한 작업을 실행하는 역할을 합니다.

3.3 Executor

각 작업(Task)을 실행하는 구성 요소로, 여러 종류가 있습니다.

• SequentialExecutor: 하나의 작업만 순차적으로 실행
• LocalExecutor: 병렬 실행 가능
• CeleryExecutor: 여러 워커(Worker) 노드에서 작업을 분산 실행
• KubernetesExecutor: 쿠버네티스를 이용한 확장성 높은 실행

3.4 Metadata Database

DAG과 작업 상태를 저장하는 데이터베이스로, 일반적으로 PostgreSQL 또는 MySQL을 사용합니다.

 

4. Apache Airflow 설치 및 실행

4.1 설치 방법

Airflow는 Python 패키지로 제공되며, pip을 이용해 쉽게 설치할 수 있습니다.

pip install apache-airflow

또는 특정 버전과 함께 설치하려면:

pip install apache-airflow==2.6.0

4.2 초기화 및 실행

airflow db init  # 데이터베이스 초기화
airflow webserver --port 8080  # 웹 서버 실행
airflow scheduler  # 스케줄러 실행

이후 http://localhost:8080에서 Airflow 웹 UI에 접속할 수 있습니다.

 

5. DAG 작성 예제

Airflow에서 DAG을 작성하는 기본적인 예제는 다음과 같습니다.

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

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

with DAG(
    'hello_airflow',
    default_args={
        'owner': 'airflow',
        'depends_on_past': False,
        'start_date': datetime(2024, 4, 1),
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
    schedule_interval='@daily',
    catchup=False,
) as dag:

    task1 = PythonOperator(
        task_id='print_hello',
        python_callable=print_hello,
    )

    task1

 

6. Airflow의 활용 사례

6.1 데이터 엔지니어링 파이프라인 구축

Airflow는 ETL(Extract, Transform, Load) 작업을 자동화하는 데 자주 사용됩니다. 예를 들어, 원천 데이터베이스에서 데이터를 추출하고, 변환한 후 데이터 웨어하우스에 적재하는 파이프라인을 쉽게 구성할 수 있습니다.

6.2 머신러닝 파이프라인 자동화

Airflow는 모델 학습, 검증, 배포 단계를 자동화하는 데 활용될 수 있습니다. 예를 들어, 데이터 전처리 후 모델을 학습시키고 결과를 평가하는 DAG을 작성할 수 있습니다.

6.3 클라우드 기반 워크플로우 관리

AWS, GCP, Azure와 같은 클라우드 서비스와의 연동이 용이하여, 클라우드 환경에서의 데이터 처리 및 분석 워크플로우를 효율적으로 운영할 수 있습니다.

 

7. Airflow 사용 시 고려사항

• 복잡한 DAG 관리: DAG이 많아지면 관리가 어려워질 수 있으므로, 모듈화된 설계를 고려해야 합니다.
• 성능 최적화: 작업이 많아질 경우, CeleryExecutor 또는 KubernetesExecutor를 활용하여 확장성을 확보해야 합니다.
• 보안 설정: Airflow의 기본 설정은 보안이 취약할 수 있으므로, 접근 제어 및 인증 설정을 강화해야 합니다.

 

8. 결론

Apache Airflow는 데이터 파이프라인 자동화를 위한 강력한 도구로, 다양한 산업에서 널리 활용되고 있습니다. DAG 기반의 유연한 워크플로우 관리, 다양한 실행 옵션, 그리고 강력한 확장성을 제공하여, 데이터 엔지니어링 및 머신러닝 파이프라인의 핵심 구성 요소로 자리 잡고 있습니다.
Airflow를 활용하여 효율적인 데이터 파이프라인을 구축하고, 데이터 처리의 자동화 및 최적화를 경험해 보세요!