Airflow

MLOps

 

• Machine Learning Operations
• 머신러닝(ML) 모델을 배포, 모니터링, 유지 관리하는 일련의 과정을 의미
• 목표: ML 모델의 개발 생명 주기를 자동화하고 최적화하는 것이 목표
• 필요성
  • 워크플로우 자동화: 모델 개발, 학습, 배포, 모니터링 등 전체 파이프라인 자동화
  • 안정성: 모델을 프로덕션 환경에 안정적으로 배포하고, 지속적인 모니터링을 통해 이상 상황 감지 및 대응
  • 확장성: 데이터와 트래픽에 맞게 인프라 확장
  • 모니터링: 실시간 성능 지표, 로그, 알림 통해 성능 저하, 오류 모니터링
• 장점
  1. 빠른 배포: 머신러닝 모델의 배포 과정을 자동화
  2. 협업 효율성 향상: 데이터 과학자, 개발자, 운영팀의 협업
  3. 모델 성능 향상: 모니터링&필요 시 업데이트
• vs DevOps
  • 공통점: 둘 다 소프트웨어 개발과 운영을 자동화하고 효율적으로 관리
  • 차이점
    • DevOps: 일반적인 소프트웨어 개발 및 운영을 자동화하여 CI/CD 파이프라인을 구축
    • MLOps: ML 모델의 지속적인 학습, 배포 자동화, 데이터 및 모델 관리
• MLOps 도구
  • 실험 관리: MLflow, Weights & Biases
  • 모델 배포: Docker, Kubernetes, FastAPI, TensorFlow Serving
  • 모델 모니터링: Prometheus, Grafana
  • CI/CD 파이프라인: Jenkins, GitHub Actions, GitLab CI/CD
  • 데이터 및 모델 버전 관리: DVC, MLflow, Pachyderm

 

MLOps의 생명주기

 

1. 모델 개발
   • 첫 번째 단계는 머신러닝 모델을 개발
   • 데이터 수집, 데이터 전처리, 모델 학습
2. 모델 배포
   • 모델 완성 후, 새로운 데이터에 대해 예측을 수행할 수 있도록 배포
   • 자동화 사용
3. 모니터링
   • 지속적으로 모니터링을 통해 모델이 올바르게 작동하는지, 데이터가 크게 변하지 않았는지 확인
4. 모델 유지보수
   • 새로운 데이터가 수집 -> 모델을 업데이트하거나 재학습

 

Airflow

 

• 데이터 엔지니어링 분야에서 가장 널리 사용되는 워크플로우 관리 도구
• Airbnb가 내부 데이터 엔지니어링 워크플로우를 자동화하기 위해 Python 기반으로 개발한 워크플로우 오케스트레이션 도구
• Flask 기반으로 작성된 태스크들(데이터셋 생성, 모델 학습 등)을 일련의 그래프로 연결하고 스케줄링, 모니터링 등 파이프라인 관리
• DAG(Directed Acyclic Graph) 개념을 사용하여 ETL 작업을 스케줄링, 모니터링, 실행

 

Airflow에서의 ETL 

 

1. 데이터 수집 (Extract)

• 다양한 소스 활용: 사용자 로그, 예약 정보, 결제 데이터, 고객 리뷰 등 여러 데이터 소스로부터 데이터를 추출
• MySQL, PostgreSQL, Google BigQuery, AWS S3 

2. 데이터 변환 (Transform)

• 정제 및 가공: 원시 데이터에는 노이즈나 결측치가 있을 수 있으므로, Pandas와 같은 도구를 활용해 데이터를 정제하고, 포맷 맞춤
• 피처 엔지니어링: 머신러닝 모델 학습에 적합하도록 데이터를 변환하고, 새로운 파생 변수를 생성
• 복잡한 처리: 로그 데이터를 기반으로 사용자 행동 패턴을 분석하거나, 예약 데이터를 통합하는 등 복잡한 데이터 변환 작업

3. 데이터 적재 (Load)

• 분석용 데이터 웨어하우스 구축: 정제된 데이터를 Snowflake, Redshift, BigQuery 등 대규모 데이터 웨어하우스에 로드
• 대시보드 및 리포팅 활용: 적재된 데이터를 Tableau, Superset 등의 BI 도구에서 활용하여, 실시간 대시보드나 리포트를 만들어 경영진과 팀원들이 인사이트를 도출할 수 있도록 지원

 

• ETL vs ELT
  
  • ETL: 전통적인 방식, 데이터를 추출하여 변환한 후 적재하는 방식
  • ELT: 먼저 데이터를 추출하여 적재한 후, 필요에 따라 변환하는 현대적 방식

 

ETL 파이프라인에서 데이터를 업데이트하는 접근 방식

 

• 데이터가 작을 경우 Full Refresh(통째로 업데이트)
  • 테이블을 완전히 비우고(Truncate) 원본 데이터를 모두 적재하는 작업(Task)을 DAG로 구성
• 양이 많을 경우 Incremental update(변경된 것만 업데이트)
  • 데이터 소스에서 마지막 업데이트 시간을 기록하거나, 로그 테이블 또는 CDC(Change Data Capture)를 활용하여 변경된 데이터만 추출
  • 이를 기반으로, 기존 데이터에 신규 데이터를 추가하거나 변경된 부분만 업데이트하는 작업(Task)을 DAG로 구성
•  Airflow 자체는 ETL 작업의 스케줄링 및 오케스트레이션을 담당
• Full Refresh와 Incremental Update 전략은 사용자가 작성하는 태스크 내에서 구현
• Airflow DAG 내에 있는 각 태스크(예: PythonOperator, SQLExecuteQueryOperator 등)를 사용해 어떤 방식으로 데이터를 로드할지 결정하고, 그 작업을 실행하도록 구성

 

멱등성

 

• 특정 연산을 여러 번 수행해도 결과가 동일하게 유지되는
• 멱등성 보장: 트랜잭션이 예시
• 멱등성이 보장되어야 테스크 실행과 관계없이 데이터가 일정하게 유지되어야 함

✅ 멱등성이 보장되는 메서드

• GET: 같은 요청을 여러 번 보내도 동일한 응답이 반환됨 (조회 연산)
• PUT: 동일한 리소스에 같은 데이터를 여러 번 업데이트해도 결과가 변하지 않음
• DELETE: 같은 리소스를 여러 번 삭제 요청해도 최종 결과는 "삭제됨" 상태로 유지

❌ 멱등성이 보장되지 않는 메서드

• POST: 같은 요청을 여러 번 보내면 새로운 리소스가 여러 개 생성될 수 있음
• PATCH: 일부 필드만 변경하므로, 연속 요청 시 결과가 다를 가능성이 있음

 

배치 프로세스와 실시간 프로세스 차이

 

배치 프로세스 

• 특징:정해진 주기(예: 하루, 매시간, 매분) 또는 조건에 따라 데이터를 한 번에 처리
• 예시: 리눅스의 Crontab, 스크립트를 주기적으로 실행하는 방식
• 문제점
  • 재실행/재시도 기능 부재: 오류 발생 시 자동으로 재실행하거나 재시도를 수행하지 않는다.
  • 알람 및 모니터링 부재: 작업 실패 시 별도의 알람이나 로그 확인 기능이 없어서 문제가 발생해도 인지하기 어렵다.
  • 설정 및 변수 관리의 어려움: 복잡한 파라미터나 설정값을 외부에서 분리하여 관리하기 어렵다. 

실시간 프로세스 

• 특징: 데이터가 발생하는 즉시 처리하는 방식으로, 지연 없이 실시간에 가까운 결과를 제공
• 예시: 스트리밍 데이터 처리, 실시간 이벤트 처리 
• 문제점: 실시간 처리는 고도의 인프라와 복잡한 처리 로직이 필요하며, 오류 발생 시 빠르게 복구 필요

 

Airflow 의 장점

• 자동 재시도 및 실패 처리: 작업이 실패하면 자동으로 재시도하고, 실패가 지속될 경우 알람
• 실행 로그 및 모니터링: 모든 작업의 로그를 중앙에서 관리하고, 웹 UI를 통해 모니터링
• 동시 실행 워커 관리: 작업 간의 의존성을 정의하고, 워커 수나 리소스 할당을 조정하여 효율적으로 동시 실행 관리
• 설정 및 변수 관리: 외부 변수와 설정 파일을 분리하여 관리할 수 있어, 환경별 설정 변경 용이

 

 

XCom

 

• Airflow의 Task 간 데이터 공유 메커니즘
• 즉, 하나의 Task에서 실행된 결과를 다른 Task에서 받아서 사용할 수 있도록 해주는 기능

 

사용 이유

 

DAG 내에서 여러 Task가 실행될 때

• Task A가 생성한 데이터를 B에서 사용해야 하는 경우
• Task A가 실행된 후, 그 결과에 따라 B의 동작을 결정해야 하는 경우
• Task간 데이터를 전달하기 위해 사용

 

기능

 

1. 값 저장 (push)
   • Task가 실행된 후, return 값을 자동으로 XCom에 저장 (implicit push)
   • ti.xcom_push(key, value)를 사용해 수동으로 값 저장 (explicit push)
2. 값 조회 (pull)
   • ti.xcom_pull(task_ids='task_A')로 특정 Task의 출력을 가져올 수 있음

 

사용할 상황

 

• 데이터 다운로드 & 처리: Task A에서 데이터를 가져오고, Task B에서 이를 변환
• API 호출 & 응답 처리: Task A에서 API 호출 후 응답을 저장하고, Task B에서 응답을 분석
• 모델 학습 & 결과 저장: Task A에서 모델을 학습하고, Task B에서 결과를 평가

 

dag와 task 간단한 예시

 

DAG 1: 데이터 수집
├── Task 1: API에서 데이터 가져오기
├── Task 2: 데이터 정리
└── Task 3: 데이터 저장

DAG 2: 모델 학습 및 배포
├── Task 1: 데이터 불러오기
├── Task 2: 모델 학습
├── Task 3: 모델 배포

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

def task_1():
    print("Task 1 실행")

def task_2():
    print("Task 2 실행")

with DAG(
    "example_dag",
    start_date=datetime(2025, 3, 19),
    schedule_interval="@daily",
    catchup=False
) as dag:
    t1 = PythonOperator(
        task_id="task_1",
        python_callable=task_1
    )

    t2 = PythonOperator(
        task_id="task_2",
        python_callable=task_2
    )

    t1 >> t2  # Task 실행 순서 지정 (t1 실행 후 t2 실행)

 

실습

from datetime import datetime, timedelta
import os
import requests
from dotenv import load_dotenv

from airflow import DAG
from airflow.operators.python import PythonOperator

# .env 파일에서 OPENWEATHER_API_KEY를 가져옴
load_dotenv()
API_KEY = os.getenv("OPENWEATHER_API_KEY")

# 도시 및 API 호출 URL
CITY = "Seoul"
URL = f"https://api.openweathermap.org/data/2.5/weather?q={CITY}&appid={API_KEY}&units=metric"


def get_weather():
    """OpenWeather API를 호출하여 서울의 날씨 데이터를 가져옴"""
    response = requests.get(URL)
    if response.status_code == 200:
        data = response.json()
        temp = data["main"]["temp"]
        weather = data["weather"][0]["description"]
        humidity = data["main"]["humidity"]
        return f"서울의 현재 날씨: {weather}, 온도: {temp}°C, 습도: {humidity}%"
    else:
        return "날씨 정보를 가져오는 데 실패했습니다."


def append_weather_to_file():
    """data 폴더의 weather.txt 파일에 실행 시각과 함께 날씨 정보를 추가함"""
    weather_info = get_weather()
    now_str = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"{now_str} (UTC) -> {weather_info}\n"

    # data 폴더가 없으면 생성
    data_dir = os.path.join(os.getcwd(), 'data')
    os.makedirs(data_dir, exist_ok=True)

    # weather.txt 파일에 append 모드로 날씨 정보 기록
    file_path = os.path.join(data_dir, "weather.txt")
    with open(file_path, "a", encoding="utf-8") as file:
        file.write(log_entry)

    print(f"{now_str}에 날씨 정보를 weather.txt에 추가했습니다.")


# 기본 DAG 인자 설정
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의: 1분마다 실행
dag = DAG(
    "weather_to_file_dag",
    default_args=default_args,
    description="DAG to append weather info to a single file every minute",
    schedule_interval=timedelta(minutes=1),
    catchup=False,
    tags=['weather', 'save'],
)

# PythonOperator: append_weather_to_file 함수를 실행하여 weather.txt 파일에 데이터를 추가
append_weather_task = PythonOperator(
    task_id="append_weather_task",
    python_callable=append_weather_to_file,
    dag=dag,
)

append_weather_task

 

실행 결과

2025-03-18 03:01:18 (UTC) ->
AAPL 최신 가격: 213.9225 USD (시간: 2025-03-17 19:59:00)
GOOGL 최신 가격: 164.4200 USD (시간: 2025-03-17 19:59:00)
AMZN 최신 가격: 196.0600 USD (시간: 2025-03-17 19:59:00)

 

from datetime import datetime, timedelta
import os
import requests
from dotenv import load_dotenv

from airflow import DAG
from airflow.operators.python import PythonOperator

# .env 파일에서 OPENWEATHER_API_KEY를 가져옴
load_dotenv()
API_KEY = os.getenv("OPENWEATHER_API_KEY")

# 도시 및 API 호출 URL
CITY = "Seoul"
URL = f"https://api.openweathermap.org/data/2.5/weather?q={CITY}&appid={API_KEY}&units=metric"


def get_weather():
    """OpenWeather API를 호출하여 서울의 날씨 데이터를 가져옴"""
    response = requests.get(URL)
    if response.status_code == 200:
        data = response.json()
        temp = data["main"]["temp"]
        weather = data["weather"][0]["description"]
        humidity = data["main"]["humidity"]
        return f"서울의 현재 날씨: {weather}, 온도: {temp}°C, 습도: {humidity}%"
    else:
        return "날씨 정보를 가져오는 데 실패했습니다."


def append_weather_to_file():
    """data 폴더의 weather.txt 파일에 실행 시각과 함께 날씨 정보를 추가함"""
    weather_info = get_weather()
    now_str = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"{now_str} (UTC) -> {weather_info}\n"

    # data 폴더가 없으면 생성
    data_dir = os.path.join(os.getcwd(), 'data')
    os.makedirs(data_dir, exist_ok=True)

    # weather.txt 파일에 append 모드로 날씨 정보 기록
    file_path = os.path.join(data_dir, "weather.txt")
    with open(file_path, "a", encoding="utf-8") as file:
        file.write(log_entry)

    print(f"{now_str}에 날씨 정보를 weather.txt에 추가했습니다.")


# 기본 DAG 인자 설정
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의: 1분마다 실행
dag = DAG(
    "weather_to_file_dag",
    default_args=default_args,
    description="DAG to append weather info to a single file every minute",
    schedule_interval=timedelta(minutes=1),
    catchup=False,
    tags=['weather', 'save'],
)

# PythonOperator: append_weather_to_file 함수를 실행하여 weather.txt 파일에 데이터를 추가
append_weather_task = PythonOperator(
    task_id="append_weather_task",
    python_callable=append_weather_to_file,
    dag=dag,
)

append_weather_task

 

실행 결과

2025-03-18 03:12:01 (UTC) -> 서울의 현재 날씨: light snow, 온도: 1.76°C, 습도: 87%
2025-03-18 03:13:01 (UTC) -> 서울의 현재 날씨: light snow, 온도: 1.43°C, 습도: 86%
2025-03-18 03:30:01 (UTC) -> 서울의 현재 날씨: broken clouds, 온도: 2.76°C, 습도: 81%
2025-03-18 03:41:01 (UTC) -> 서울의 현재 날씨: mist, 온도: 2.76°C, 습도: 81%