Apache Airflow로 데이터 파이프라인 구축하기 ETL 프로세스 자동화

Apache Airflow로 데이터 파이프라인 구축하기: ETL 프로세스 자동화

Overview

Apache Airflow는 복잡한 데이터 파이프라인을 구축하고 관리하는 데 매우 유용한 오픈소스 도구입니다. 이 글에서는 Apache Airflow를 사용하여 ETL(Extract, Transform, Load) 프로세스를 자동화하는 방법에 대해 자세히 설명하겠습니다. ETL은 데이터의 추출, 변환, 적재 과정을 의미하며, 데이터 분석, 머신러닝 및 비즈니스 인사이트 도출에 중요한 역할을 합니다. Airflow는 DAG(Directed Acyclic Graph) 개념을 기반으로 작업을 정의하고 실행하는 강력한 기능을 제공합니다.

Apache Airflow 개요

Apache Airflow는 데이터 파이프라인을 구성하는 데 필요한 다양한 기능을 제공합니다. 주요 기능은 다음과 같습니다:

1. DAG(Directed Acyclic Graph): 각 작업의 의존성과 실행 순서를 정의합니다.
2. Operator: 데이터 작업을 정의하는 클래스입니다. 예를 들어, BashOperator, PythonOperator, PostgresOperator 등이 있습니다.
3. Scheduler: DAG를 주기적으로 실행하고 관리합니다.
4. Web UI: 파이프라인의 상태를 모니터링하고 관리하는 웹 인터페이스를 제공합니다.

이제 Airflow를 사용하여 ETL 프로세스를 구현하는 방법을 단계별로 알아보겠습니다.

단계 1: Apache Airflow 설치

먼저 Apache Airflow를 설치해야 합니다. Airflow는 pip를 통해 간단히 설치할 수 있습니다.

# 가상환경 생성 및 활성화
python3 -m venv airflow_venv
source airflow_venv/bin/activate

# Apache Airflow 설치
export AIRFLOW_VERSION=2.7.0
export PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
export CONSTRAINT_FILE="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

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

이렇게 설치한 후, Airflow를 초기화하고 시작합니다.

# 데이터베이스 초기화
airflow db init

# Airflow 웹 서버 시작
airflow webserver --port 8080

# Airflow 스케줄러 시작
airflow scheduler

단계 2: ETL 프로세스 정의하기

이제 ETL 프로세스를 정의할 차례입니다. 예를 들어, CSV 파일에서 데이터를 추출하고, 이를 변환한 후 데이터베이스에 적재하는 과정을 구현해 보겠습니다.

1. Extract: CSV 파일에서 데이터를 읽습니다.
2. Transform: 데이터를 가공합니다. 예를 들어, 결측치를 처리하거나 데이터를 정규화하는 과정입니다.
3. Load: 변환된 데이터를 데이터베이스에 적재합니다.

DAG 파일 생성

Airflow의 DAG는 Python 파일로 정의됩니다. 아래는 etl_pipeline.py라는 파일에 작성할 DAG 예시입니다.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
import pandas as pd
import sqlite3

# ETL 함수 정의
def extract():
df = pd.read_csv('data.csv')  # CSV 파일에서 데이터 추출
return df

def transform(df):
# 결측치 처리
df.fillna(0, inplace=True)
# 추가적인 변환 로직 구현
return df

def load(df):
# SQLite 데이터베이스에 적재
conn = sqlite3.connect('example.db')
df.to_sql('data_table', conn, if_exists='replace', index=False)
conn.close()

# DAG 정의
with DAG(
dag_id='etl_pipeline',
schedule_interval='@daily',
start_date=days_ago(1),
catchup=False,
) as dag:

extract_task = PythonOperator(
task_id='extract',
python_callable=extract,
)

transform_task = PythonOperator(
task_id='transform',
python_callable=lambda: transform(extract_task.output),
)

load_task = PythonOperator(
task_id='load',
python_callable=lambda: load(transform_task.output),
)

extract_task >> transform_task >> load_task  # 작업 간의 의존성 설정

단계 3: DAG 실행 및 모니터링

DAG 파일을 dags 폴더에 저장한 후, Airflow 웹 UI에 접속하여 DAG를 실행할 수 있습니다. 웹 UI에서는 DAG의 상태, 실행 로그 등을 확인할 수 있습니다.

• DAG 상태 모니터링: 각 작업의 성공 여부를 실시간으로 확인할 수 있습니다.
• 로그 확인: 특정 작업을 클릭하여 실행 로그를 확인하고, 발생한 에러를 진단할 수 있습니다.

에러 발생 시 대처 방법

작업 실행 중 오류가 발생할 수 있습니다. 예를 들어, CSV 파일 경로가 잘못되었거나, 데이터베이스 연결이 실패하는 경우입니다. 에러 메시지를 통해 원인을 파악하고, 아래와 같은 방식으로 해결할 수 있습니다.

FileNotFoundError: [Errno 2] No such file or directory: 'data.csv'

이러한 에러는 파일 경로를 확인하고, 필요한 파일이 존재하는지 검토함으로써 해결할 수 있습니다.

결론

Apache Airflow를 사용하면 ETL 프로세스를 효율적으로 자동화할 수 있습니다. DAG 구조를 통해 작업의 의존성을 명확히 하여, 데이터 파이프라인을 쉽게 관리하고 모니터링할 수 있습니다. 이 글에서는 기본적인 ETL 예제를 통해 Airflow의 사용법을 소개하였으나, 다양한 Operator와 Custom Operator를 통해 더욱 복잡한 데이터 파이프라인을 구축할 수 있습니다.

참고문서

• Apache Airflow Documentation
• Airflow GitHub Repository
• Airflow Operators