🚀 데이터 파이프라인 자동화의 끝판왕, Airflow 파헤치기!

이번 포스팅은 데이터 엔지니어링 여정에 강력한 날개를 달아줄 Airflow에 대한 모든 것을 담았다. 긴 스크롤 압박이 예상되지만, 그만큼 알찬 정보로 가득 채우려고 노력했다! 🔥

1. 서론: 왜 Airflow가 필요할까? 🤔

LLM 학습에 빠져 살고 있는 주니어 개발자로서, 데이터를 처리하는 과정에서 수많은 어려움을 겪었다. 수동으로 데이터를 옮기고, 스크립트를 실행하고, 오류를 감시하는 일은 정말 고됐다. 😵‍💫

이런 복잡하고 반복적인 작업을 자동화할 수 있는 방법이 없을까? 고민하던 중 눈에 들어온 것이 바로 Airflow!

Airflow는 복잡한 데이터 파이프라인을 정의, 스케줄링, 모니터링할 수 있는 강력한 오픈소스 플랫폼이다. 데이터 엔지니어링의 필수 도구로 자리매김하고 있으며 데이터 처리 자동화의 핵심 기술이라고 할 수 있음 !!

Airflow가 필요한 이유:

• 수동 작업의 한계: 데이터 처리 작업은 반복적이고 오류가 발생하기 쉽다. Airflow는 이러한 수동 작업을 자동화하여 효율성을 높여준다.
• 복잡한 의존성 관리: 여러 작업이 복잡하게 연결된 파이프라인을 관리하는 것은 매우 어렵다. Airflow는 작업 간 의존성을 명확하게 정의하고 관리할 수 있도록 도와준다.
• 실패 처리 및 재시도: 작업 실패 시 자동 재시도 기능을 제공하여 데이터 처리 안정성을 높여준다.
• 모니터링 및 알림: 파이프라인의 실행 상태를 모니터링하고, 오류 발생 시 알림을 받을 수 있어 문제 발생 시 빠르게 대응할 수 있다.

---

2. Airflow 핵심 개념 짚어보기 🧐

Airflow를 제대로 활용하기 위해서는 몇 가지 핵심 개념을 이해해야 한다. 마치 레고 블록처럼, 이 개념들을 조합하여 데이터 파이프라인을 만들 수 있는 것 !

• DAG (Directed Acyclic Graph): Airflow의 핵심 단위로, 작업(task)들의 의존 관계를 방향성 비순환 그래프(DAG) 형태로 표현. DAG는 파이프라인의 논리적 구조를 정의.
• Task: DAG를 구성하는 개별 작업 단위를 의미. (데이터 추출, 변환, 로드 등)
• Operator: Task가 수행하는 실제 작업을 정의. Airflow는 다양한 종류의 Operator를 제공하며, 사용자 정의 Operator를 만들 수도 있음. (BashOperator, PythonOperator, PostgresOperator 등)
• Scheduler: 정의된 DAG를 스케줄에 따라 실행하는 역할. Cron 표현식을 사용하여 실행 주기를 설정할 수 있다.
• Executor: Task를 실행하는 엔진. (LocalExecutor, CeleryExecutor, KubernetesExecutor 등)
• Web UI: Airflow 웹 인터페이스를 통해 DAG 실행 상태를 모니터링하고, 로그를 확인하고, 스케줄을 관리

---

3. Airflow, 이렇게 활용하세요! ✨

Airflow는 데이터 엔지니어링의 다양한 분야에서 활용될 수 있다. 몇 가지 예시를 통해 Airflow의 활용 가능성을 살펴보자 !

• 데이터 수집 및 통합: API, 데이터베이스, 파일 시스템 등 다양한 소스에서 데이터를 수집하고 통합하는 파이프라인 구축
• 데이터 변환 및 정제: 수집된 데이터를 원하는 형태로 변환하고 정제하는 파이프라인을 구축할 수 있다. (데이터 누락 값 처리, 이상치 제거, 데이터 형식 변환 등을 자동화)
• 머신러닝 모델 학습: 모델 학습에 필요한 데이터 전처리 과정을 자동화하고, 모델 학습 및 배포 파이프라인 구축
• 데이터 웨어하우스 구축: 데이터 웨어하우스에 데이터를 적재하고 관리하는 파이프라인 구축
• BI (Business Intelligence) 보고서 생성: 데이터를 분석하여 BI 보고서를 생성하고 배포하는 파이프라인 구축

Airflow 활용 예시 (Python 코드):

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from datetime import datetime

def my_python_function():
    print("Hello from Python!")

with DAG(
    dag_id='my_first_airflow_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    bash_task = BashOperator(
        task_id='run_bash_command',
        bash_command='echo "Hello from Bash!"'
    )

    python_task = PythonOperator(
        task_id='run_python_function',
        python_callable=my_python_function
    )

    bash_task >> python_task

위 코드는 Bash 명령어를 실행하는 Task와 Python 함수를 실행하는 Task로 구성된 간단한 DAG다. 이 코드를 Airflow에 등록하면, 정의된 순서대로 Task가 실행된다.

---

4. airflow 설치와 사용

1) 설치

pip install apache-airflow
이렇게 pip로 직접 설치하거나

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'
airflow에서 제공하는 docker-compose 파일을 받아와도 된다.

이런거 너무 귀여움

나의 경우 기존에 사용하던 MySQL에서 PostgreSQL로 이동하는 코드를 작성했다.

• compose file(airflow 기본 세팅)
  이 부분에서 데이터베이스를 설치하고 로그인 정보를 입력하면 된다.

  version: '3.7'
  services:
    postgres:
      image: postgres:13
      environment:
        POSTGRES_USER: airflow
        POSTGRES_PASSWORD: airflow
        POSTGRES_DB: airflow
      ports:
        - "5432:5432"
      volumes:
        - postgres_data:/var/lib/postgresql/data
  
    mysql:
      image: mysql:8.0
      environment:
        MYSQL_ROOT_PASSWORD: root
        MYSQL_DATABASE: source
        MYSQL_USER: user
        MYSQL_PASSWORD: password
      ports:
        - "3306:3306"
      volumes:
        - mysql_data:/var/lib/mysql
  
    redis:
      image: redis:7.2-bookworm
      ports:
        - "6379:6379"
  
    airflow-webserver:
      image: apache/airflow:2.7.1
      environment:
        - AIRFLOW__CORE__EXECUTOR=LocalExecutor
        - AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
        - AIRFLOW__CELERY__BROKER_URL=redis://redis:6379/0
      depends_on:
        - postgres
        - mysql
        - redis
      ports:
        - "8080:8080"
      volumes:
        - ./dags:/opt/airflow/dags
        - ./logs:/opt/airflow/logs
        - ./plugins:/opt/airflow/plugins
      command: >
        bash -c "
        airflow db init &&
        airflow users create --username airflow --password airflow --firstname Admin --lastname User --role Admin --email admin@example.com &&
        airflow webserver"
  
    airflow-scheduler:
      image: apache/airflow:2.7.1
      depends_on:
        - postgres
        - redis
      volumes:
        - ./dags:/opt/airflow/dags
        - ./logs:/opt/airflow/logs
        - ./plugins:/opt/airflow/plugins
      command: >
        bash -c "
        airflow scheduler"
  
  volumes:
    postgres_data:
    mysql_data:

---

그 다음 docker-compose up -d로 올리면 기본 세팅 끝 !

• 마이그레이션 작업할 py 파일

  from airflow import DAG
  from airflow.operators.python import PythonOperator
  from datetime import datetime, timedelta
  import mysql.connector
  import psycopg2
  # MySQL에서 데이터 가져오기
  def transfer_data():
      # MySQL 연결
      mysql_conn = mysql.connector.connect(
          host="mysql",
          user="user",
          password="password",
          database="db"
      )
      mysql_cursor = mysql_conn.cursor()
      mysql_cursor.execute("SELECT * FROM your_table LIMIT 100")
      rows = mysql_cursor.fetchall()
  
      # PostgreSQL 연결
      postgres_conn = psycopg2.connect(
          host="postgres",
          user="airflow",
          password="airflow",
          database="airflow"
      )
      postgres_cursor = postgres_conn.cursor()
  
      # PostgreSQL에 데이터 삽입
      insert_query = """
      INSERT INTO my_table (job_title, job_post, company)
      VALUES (%s, %s, %s)
      """
      postgres_cursor.executemany(insert_query, rows)
  
      # 커밋 및 연결 종료
      postgres_conn.commit()
      mysql_cursor.close()
      mysql_conn.close()
      postgres_cursor.close()
      postgres_conn.close()
  # DAG 정의
  default_args = {
      "owner": "airflow",
      "depends_on_past": False,
      "email_on_failure": False,
      "email_on_retry": False,
      "retries": 1,
      "retry_delay": timedelta(minutes=5),
  }
  
  with DAG(
      "mysql_to_postgres",
      default_args=default_args,
      description="Transfer data from MySQL to PostgreSQL",
      schedule_interval="0 * * * *",  # 매시간 0분에 실행
      start_date=datetime(2025, 1, 1),
      catchup=False,
  ) as dag:
  
      transfer_task = PythonOperator(
          task_id="transfer_data",
          python_callable=transfer_data,
      )

매 시간마다 100개씩 옮기도록 설정해두었다.
yaml파일에서 설정한 http://localhost:8080로 접속해 로그인하면 배치 작업을 확인할 수 있다.

이런식으로 !
간단한 작업이어서 나도 조금 더 써봐야 알겠지만
활용도가 높은 툴이다보니 더 깊게 공부해봐야겠다는 생각 !!
나중에는 데이터 분석을 위한 데이터 파이프라인도 만들어봐야겠다.

---

5. 결론: 데이터 엔지니어링 여정의 든든한 동반자, Airflow! 🤝

Airflow는 데이터 파이프라인 자동화를 위한 강력한 도구다. 복잡한 데이터 처리 작업을 자동화하고, 효율성을 높이며, 안정적인 시스템을 구축할 수 있도록 도와준다. 주니어 입장에서 Airflow를 배우고 활용하는 것은 데이터 엔지니어링 분야의 성장에 큰 도움이 될 것이다.

나도 아직 Airflow를 배우는 과정에 있지만, 앞으로도 Airflow를 활용한 데이터 파이프라인 구축 경험을 공유하며 계속 성장해 나가야겠다! 🚀

---

마무리:

이 포스팅이 Airflow에 대한 여러분의 이해를 높이는 데 도움이 되었기를 바랍니다. 궁금한 점이 있다면 언제든지 댓글로 질문해주세요. 데이터 엔지니어링의 세계에서 함께 성장해 나가는 멋진 여정을 기대합니다! 💪