데이터로 보는 세상

데이터 엔지니어링 분야에서 Apache Airflow는 데이터 파이프라인의 오케스트레이션을 위한 필수 도구로 자리잡고 있습니다. 이 블로그 글에서는 데이터 엔지니어를 위한 Airflow의 활용법을 구체적으로 살펴보겠습니다. Airflow의 기본 개념부터 시작해, DAG(Directed Acyclic Graph) 설정, 태스크 구성, 스케줄링, 그리고 모니터링에 이르기까지 실용적인 팁을 제공하겠습니다.

Airflow 개요

Apache Airflow는 복잡한 데이터 파이프라인을 효율적으로 관리하고 자동화하기 위한 플랫폼입니다. 다른 데이터 처리 도구들과의 통합이 용이하고, 웹 기반 UI를 통해 작업의 상태를 쉽게 모니터링할 수 있는 장점이 있습니다. Airflow의 핵심은 DAG입니다. DAG는 작업 간의 의존성을 정의하여, 작업이 순차적으로 또는 병렬로 실행될 수 있도록 합니다.

Airflow 설치

Airflow를 설치하기 위해서는 Python과 pip가 설치되어 있어야 합니다. 다음 명령어를 사용하여 Airflow를 설치합니다.

export AIRFLOW_VERSION=2.5.0
export PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
export CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

DAG 설정

DAG를 정의하려면 Python 파일을 작성해야 합니다. 아래는 간단한 DAG의 예제입니다. 이 DAG는 매일 오전 6시에 데이터를 가져오고, 처리하며, 결과를 저장하는 태스크를 포함하고 있습니다.

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

def extract_data():
    print("데이터를 추출합니다.")

def transform_data():
    print("데이터를 변환합니다.")

def load_data():
    print("데이터를 적재합니다.")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
}

with DAG('etl_process', default_args=default_args, schedule_interval='@daily') as dag:
    start_task = DummyOperator(task_id='start')
    
    extract_task = PythonOperator(task_id='extract', python_callable=extract_data)
    transform_task = PythonOperator(task_id='transform', python_callable=transform_data)
    load_task = PythonOperator(task_id='load', python_callable=load_data)

    start_task >> extract_task >> transform_task >> load_task

위 코드는 etl_process라는 이름의 DAG을 생성합니다. DummyOperator를 사용하여 DAG의 시작점을 설정하고, PythonOperator를 통해 각각의 ETL 태스크를 정의합니다. 태스크 간의 의존성은 >> 연산자를 통해 명시합니다.

태스크 구성

Airflow의 유연한 태스크 구성 기능을 활용하면 다양한 작업을 쉽게 자동화할 수 있습니다. 예를 들어, BashOperator를 사용하면 시스템 명령어를 실행할 수 있습니다.

from airflow.operators.bash_operator import BashOperator

bash_task = BashOperator(
    task_id='run_bash_command',
    bash_command='echo "Hello, Airflow!"',
    dag=dag,
)

스케줄링

Airflow는 다양한 스케줄링 옵션을 제공합니다. schedule_interval 파라미터를 사용하여 DAG의 실행 주기를 설정할 수 있습니다. 예를 들어, 매일, 매시간, 특정 요일에 따라 스케줄을 조정할 수 있습니다.

아래 예제는 매주 월요일 오전 6시에 실행되도록 설정하는 방법입니다.

with DAG('weekly_dag', schedule_interval='0 6   1') as dag:
    # 태스크 정의

모니터링

Airflow는 웹 기반 UI를 통해 DAG의 실행 상태와 성과를 모니터링할 수 있습니다. UI에서는 DAG의 실행 이력, 태스크의 성공 및 실패 여부, 로그 등의 정보를 확인할 수 있습니다. 이를 통해 문제 발생 시 신속하게 대응할 수 있습니다.

마무리

Apache Airflow는 복잡한 데이터 파이프라인을 효율적으로 관리하고 자동화하기 위한 강력한 도구입니다. DAG 설정, 태스크 구성 및 스케줄링 등 다양한 기능을 활용하여 데이터 엔지니어링 작업을 간소화할 수 있습니다. 본 글에서 소개한 내용을 바탕으로 Airflow를 실무에 적용해보시기 바랍니다. 데이터 환경이 복잡해질수록 Airflow의 필요성은 더욱 커질 것입니다.