[데이터 TIL] Apache Airflow란?

Apache Airflow: 여러 데이터 소스를 활용한 데이터 분석 자동화

 

📌 Apache Airflow란?

Apache Airflow는 워크플로우(Workflow) 자동화 및 오케스트레이션 도구로, 데이터 수집, 변환(ETL), 분석 및 배포 과정을 효율적으로 관리할 수 있도록 도와준다.

Airflow는 DAG(Directed Acyclic Graph, 방향 비순환 그래프)라는 구조를 사용하여 각 데이터 처리 단계를 정의하고 실행 순서를 설정할 수 있도록 설계되었다.

 

💡 쉽게 말해?

✔ 여러 데이터 소스에서 데이터를 자동으로 수집하고, 변환하고, 분석할 수 있도록 도와주는 도구!

✔ 복잡한 데이터 파이프라인을 자동화하고, 일정에 맞춰 실행하며, 오류가 발생하면 다시 시도할 수 있도록 설계됨

 

---

 

📌 Apache Airflow를 사용하는 이유

 

✅ 여러 데이터 소스를 통합하여 분석용 데이터 만들기

✅ ETL(Extract, Transform, Load) 파이프라인 자동화

✅ 데이터 파이프라인의 실행 일정(스케줄링) 및 오류 복구 관리

✅ 대용량 데이터를 효율적으로 처리하고, 클라우드 환경과 쉽게 연동 가능

 

✔ 예제:

• 여러 데이터베이스(DB)에서 데이터를 수집 (MySQL, PostgreSQL, MongoDB 등)

• API 데이터를 주기적으로 수집 (예: 네이버 뉴스, 공공 데이터 API)

• AWS S3, Google Cloud Storage(GCS)에서 데이터 읽어오기

• Spark, Hadoop, BigQuery 등의 빅데이터 분석 시스템과 연계

• 머신러닝 모델 학습을 위한 데이터 전처리 자동화

 

반응형

---

 

📌 Airflow를 활용한 데이터 수집 및 분석 프로세스

 

Airflow는 다양한 데이터 소스에서 데이터를 가져와 분석용 데이터로 가공하는 전 과정을 자동화할 수 있다

 

🔹 1️⃣ 여러 데이터 소스에서 데이터 수집

✔ Airflow의 다양한 오퍼레이터(Operators)를 활용하여 여러 데이터 소스에서 데이터 추출 가능

✔ SQL 데이터베이스, API, 클라우드 스토리지 등에서 데이터를 가져오는 작업을 자동화

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

dag = DAG('data_ingestion',
          default_args=default_args,
          schedule_interval='@daily',
          catchup=False)

# PostgreSQL에서 데이터 가져오기
extract_postgres = PostgresOperator(
    task_id='extract_postgres',
    postgres_conn_id='postgres_default',
    sql="SELECT * FROM users;",
    dag=dag
)

# API 데이터 수집
extract_api = SimpleHttpOperator(
    task_id='extract_api',
    http_conn_id='my_api',
    endpoint='data',
    method='GET',
    dag=dag
)

# S3에서 Redshift로 데이터 로드
load_to_redshift = S3ToRedshiftOperator(
    task_id='load_s3_to_redshift',
    schema='public',
    table='analytics_table',
    s3_bucket='my-data-bucket',
    s3_key='data/file.csv',
    copy_options=['CSV'],
    redshift_conn_id='redshift_default',
    dag=dag
)

extract_postgres >> extract_api >> load_to_redshift

✔ PostgreSQL, API, S3 데이터를 연동하여 Redshift에 적재하는 데이터 파이프라인 예제

✔ DAG를 통해 데이터 추출-변환-적재(ETL) 과정 정의

 

 

🔹 2️⃣ 데이터 변환(ETL)

✔ 수집한 데이터를 정제하고 변환하여 분석용 데이터로 가공

✔ Spark, Pandas, SQL 등을 활용하여 데이터를 처리한 후 저장

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

transform_data = SparkSubmitOperator(
    task_id='transform_data',
    application='/path/to/spark_script.py',
    conn_id='spark_default',
    dag=dag
)

✔ SparkSubmitOperator를 사용하여 Spark 작업 실행

 

SMALL

 

🔹 3️⃣ 분석 및 시각화용 데이터 저장

✔ 분석을 위해 데이터 웨어하우스(BigQuery, Redshift) 또는 데이터베이스에 저장

✔ BI 도구(Tableau, Looker)와 연결하여 데이터 시각화 가능

from airflow.providers.google.cloud.transfers.bigquery_to_gcs import BigQueryToGCSOperator

export_to_gcs = BigQueryToGCSOperator(
    task_id='export_to_gcs',
    source_project_dataset_table='my_project.analytics_table',
    destination_cloud_storage_uris=['gs://my-bucket/export_data.csv'],
    dag=dag
)

✔ BigQuery 데이터를 Google Cloud Storage로 내보내는 작업

 

---

 

📌 Airflow의 주요 기능

 

✅ 1. DAG(Directed Acyclic Graph) 기반 워크플로우

• DAG를 사용하여 작업(Task) 간의 의존성(Dependency)을 정의

• 데이터 흐름을 순차적 혹은 병렬적으로 실행 가능

 

✅ 2. 강력한 스케줄링 기능

• 주기적인 데이터 수집 및 처리 자동화

• CRON 표현식(@daily, @hourly 등)으로 설정 가능

 

✅ 3. 확장성과 클라우드 연동

• AWS, GCP, Azure 등 클라우드 환경에서 Spark, Redshift, BigQuery 등과 연동 가능

• Celery Executor를 사용하여 여러 노드에서 병렬 처리 가능

 

✅ 4. 웹 기반 UI 제공

• 대시보드를 통해 DAG 실행 상태, Task 진행 상황을 한눈에 확인 가능

 

---

 

📌 Airflow의 장점과 단점

 

✅ 장점

✔ 여러 데이터 소스를 통합할 수 있음

✔ 데이터 파이프라인 자동화 & 일정 관리 가능

✔ 웹 UI를 통해 실행 상태를 시각적으로 확인 가능

✔ Python 기반으로 확장성과 커스터마이징이 뛰어남

 

⚠️ 단점

❌ 설치 및 설정이 다소 복잡할 수 있음

❌ 실시간 데이터 스트리밍 처리는 적합하지 않음 (주로 배치 처리)

 

---

 

Airflow를 사용해야 하는 이유

• 여러 데이터 소스에서 데이터를 수집하고, 변환 및 분석용 데이터로 만들고 싶다면?

→ Airflow가 가장 적합한 도구!

• 데이터 파이프라인을 자동화하고 효율적으로 관리하고 싶다면?

→ Airflow의 DAG 기반 워크플로우가 최고의 해결책!

 

Airflow는 대규모 데이터 분석, 머신러닝 모델 학습, ETL 파이프라인 자동화 등 다양한 데이터 엔지니어링 및 분석 작업에서 필수적인 도구이다