MLFlow

개념

 

MinIO란? MinIO는 AWS S3와 호환되는 오브젝트 스토리지 시스템으로, 데이터를 저장하고 관리하는 역할

 

 

실습

 

요청사항

 

1. data는 minio의 raw-data 버킷에서 가져와서 airflow로 매 시간 정각에 실행되는, 로그는 mlflow에 남는 pipeline dag를 완성
2. dag이름은 00_airflow_mlops.py로 합니다.
3. experiment 이름은 fisa-ml

 

사고과정

 

MinIO에서 데이터를 가져와서, 데이터를 처리한 후, MLflow에 로깅하는 데이터 파이프라인을 구축하는 DAG를 만들면 된다.

 

1. MinIO에서 raw-data 버킷에 있는 data.csv 파일을 가져옴
2. 가져온 데이터를 처리하여 통계값(평균)을 계산
3. MLflow에 로그를 기록하여 실험 관리
4. 매 시간 정각(0분)마다 실행되는 자동화된 데이터 파이프라인

 

1. 설정

 

from datetime import datetime, timedelta
import os
import boto3
import pandas as pd
import mlflow
from airflow import DAG
from airflow.operators.python import PythonOperator

# ------------------------------
# 1. MinIO 설정
# ------------------------------
MINIO_ENDPOINT = "http://minio-server:9000"
MINIO_ACCESS_KEY = "minio"
MINIO_SECRET_KEY = "miniostorage"
BUCKET_NAME = "raw-data"  # 사전에 생성
FILE_NAME = "data.csv"

# ------------------------------
# 2. MLflow 설정
# ------------------------------
MLFLOW_TRACKING_URI = "http://mlflow:600"
EXPERIMENT_NAME = "fisa-ml"  # MLflow에서 사용할 Experiment 이름

# ------------------------------
# 3. MinIO 클라이언트 생성
# ------------------------------
# boto3를 사용하여 MinIO에 연결
s3_client = boto3.client(
    "s3",
    endpoint_url=MINIO_ENDPOINT,
    aws_access_key_id=MINIO_ACCESS_KEY,
    aws_secret_access_key=MINIO_SECRET_KEY,
)

 

 

MinIO 설정 

 

• MinIO 서버는 http://minio-server:9000에서 실행됨
  • (localhost로 해도 되지만, 난 도커 컴포즈로 띄어서 컨테이너별로 구분하기 위해 해당 주소를 사용했다.)

왜 minio-server:9000 을 사용하는가

 

• Airflow 컨테이너에서 MinIO 컨테이너를 접근해야 하기 때문
• Docker Compose 내에서 컨테이너 간 통신을 하기 때문

정리

• 도커 컨테이너는 각자 독립된 네트워크를 사용
• localhost를 쓰면 Airflow 컨테이너 내부에서 실행되는 localhost를 가리키게 됨
• MinIO는 Airflow 내부의 localhost가 아니라, 별도의 MinIO 컨테이너에서 실행 중
• 따라서 MinIO의 컨테이너 이름(minio-server)을 사용해야 함

 

개념

• docker-compose.yml에서 같은 services 아래 정의된 컨테이너들은 자동으로 같은 네트워크에 속함
• 따라서 컨테이너 간에는 서비스명(컨테이너명)으로 접근 가능
• minio-server라는 서비스명이 컨테이너의 호스트 이름 역할을 함
• services 아래 minio-server로 정의 → 이게 컨테이너의 네트워크 이름이 됨

 

내 도커 컴포즈 minio 부분

services:
  minio-server:
    image: minio/minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: miniostorage
    command: server /data/minio --console-address :9001

 

도커로 minio는 9000, airflow 8080, mlflow 600으로 띄운 상태

• 접근을 위해 minio(아이디) / miniostorage(비밀번호)를 사용
• raw-data 버킷에서 data.csv 파일을 가져올 예정

 

MLflow 설정 

 

• MLflow 서버는 http://mlflow:600에서 실행됨
• fisa-ml이라는 Experiment에 데이터를 기록할 예정
• 로그할 데이터는 통계 정보(ex. 평균값)로 저장됨

 

2. MinIO에서 데이터 가져오기

 

def fetch_data_from_minio():

    local_file_path = f"/tmp/{FILE_NAME}"  # 데이터를 저장할 로컬 경로 지정

    # MinIO에서 파일 다운로드
    s3_client.download_file(BUCKET_NAME, FILE_NAME, local_file_path)
    print(f"📥 MinIO에서 {FILE_NAME} 다운로드 완료: {local_file_path}")

    # 다운로드된 파일 경로를 반환 (XCom을 통해 다음 Task에서 사용 가능)
    return local_file_path

 

1. MinIO에서 raw-data 버킷에 있는 data.csv 파일을 다운로드
2. 다운로드된 파일을 로컬 /tmp/data.csv 경로에 저장
3. Airflow의 XCom 기능을 활용하여 파일 경로를 다음 Task에 전달

 

http://localhost:9001/browser 으로 MinIO에 접속

raw-data라는 버킷안에 data.csv 파일이 위치해 있어야 한다.

 

3. 데이터 처리 & MLflow 기록

 

def process_data(**kwargs):

    # Airflow XCom에서 fetch_data Task가 반환한 파일 경로 가져오기
    ti = kwargs["ti"]
    file_path = ti.xcom_pull(task_ids="fetch_data")

    # CSV 파일 로드
    df = pd.read_csv(file_path)
    print(f"📊 데이터 로드 완료: {df.shape}")

    # 수치형 컬럼 평균 계산
    mean_values = df.mean().to_dict()

    # MLflow 설정
    mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
    mlflow.set_experiment(EXPERIMENT_NAME)

    # 컬럼명 변환 (공백, 괄호 등 제거)
    cleaned_mean_values = {col.replace(" ", "_").replace("(", "").replace(")", ""): val for col, val in mean_values.items()}

    # MLflow 실행
    with mlflow.start_run():
        mlflow.log_params({"source": "MinIO", "file": FILE_NAME})
        mlflow.log_metrics(cleaned_mean_values)

    print("✅ MLflow에 데이터 기록 완료!")

 

다운로드한 데이터를 처리하고, MLflow에 로깅하는 함수

1. fetch_data Task에서 다운로드한 파일의 경로를 XCom을 이용해 가져옴
2. pandas를 사용하여 CSV 데이터를 로드하고 수치형 컬럼의 평균값을 계산
3. mlflow.log_metrics(mean_values)로 MLflow에 평균값을 기록
4. 컬럼명에 공백/특수문자가 있으면 변환하여 저장 (MLflow 규칙 준수)

 

EXPERIMENT_NAME = "fisa-ml"로 설정했기에,  http://localhost:600/ 으로 접속 시 문제가 없다면 아래와 같이 MLflow가 실행 완료 된 것을 확인할 수 있다.

 

Experiments의 이름이 fisa-ml로 만들어진 것을 확인 가능

 

XCOM?

https://koreatstm.tistory.com/268#XCom-1

 

4. DAG 정의 및 실행 순서

 

default_args = {
    "owner": "fisa",
    "start_date": datetime(2025, 3, 19),
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "00_airflow_mlops",
    default_args=default_args,
    description="MinIO에서 데이터를 가져와 MLflow에 기록하는 DAG",
    schedule_interval="0 * * * *", 
    catchup=False,
    tags=["minio", "mlflow", "airflow"],
)

 

1. Task 실행 실패 시, 5분 간격으로 최대 1번 재시도
2. DAG는 매 시간 정각(0분)에 실행됨  (crontab 형식으로 정의)

 

📌 5. DAG의 Task 정의

 

fetch_data_task = PythonOperator(
    task_id="fetch_data",
    python_callable=fetch_data_from_minio,
    dag=dag,
)

process_data_task = PythonOperator(
    task_id="process_data",
    python_callable=process_data,
    provide_context=True,
    dag=dag,
)

fetch_data_task >> process_data_task

 

1. fetch_data_task → MinIO에서 데이터를 가져오는 Airflow Task
2. process_data_task → 가져온 데이터를 MLflow에 기록하는 Airflow Task
3. Task 간 실행 순서 지정 → fetch_data_task >> process_data_task

 

fetch_data, process_data Task 성공적으로 완료된 것을 확인

왼쪽 그래프의 맨 오른쪽 보고 화인 가능, 그 옆에는 시행착오 과정

 

전체 코드

from datetime import datetime, timedelta
import os
import boto3
import pandas as pd
import mlflow
from airflow import DAG
from airflow.operators.python import PythonOperator

# ------------------------------
# 1. MinIO 설정
# ------------------------------
MINIO_ENDPOINT = "http://minio-server:9000"
MINIO_ACCESS_KEY = "minio"
MINIO_SECRET_KEY = "miniostorage"
BUCKET_NAME = "raw-data"  # 사전에 생성
FILE_NAME = "data.csv"

# ------------------------------
# 2. MLflow 설정
# ------------------------------
MLFLOW_TRACKING_URI = "http://mlflow:600"
EXPERIMENT_NAME = "fisa-ml"  # MLflow에서 사용할 Experiment 이름

# ------------------------------
# 3. MinIO 클라이언트 생성
# ------------------------------
# boto3를 사용하여 MinIO에 연결
s3_client = boto3.client(
    "s3",
    endpoint_url=MINIO_ENDPOINT,
    aws_access_key_id=MINIO_ACCESS_KEY,
    aws_secret_access_key=MINIO_SECRET_KEY,
)


def fetch_data_from_minio():
    """
    MinIO에서 지정된 파일을 다운로드하여 로컬 `/tmp` 디렉토리에 저장하는 함수
    /tmp는 Airflow가 실행중인 서버의 임시 디렉토리
    이후 XCOM을 사용하여 다음 Task에서 이 파일 사용 가능
    """
    local_file_path = f"/tmp/{FILE_NAME}"  # 파일을 저장할 경로 설정

    # MinIO에서 파일 다운로드
    s3_client.download_file(BUCKET_NAME, FILE_NAME, local_file_path)
    print(f"📥 MinIO에서 {FILE_NAME} 다운로드 완료: {local_file_path}")

    # 다운로드된 파일 경로를 반환
    return local_file_path

def process_data(**kwargs):
    """
    다운로드한 데이터를 처리하고, MLflow에 로깅하는 함수
    """
    # Airflow XCom을 사용하여 이전 task(fetch_data)의 출력 값 가져오기
    ti = kwargs["ti"]
    file_path = ti.xcom_pull(task_ids="fetch_data")  # fetch_data에서 반환된 파일 경로 가져옴

    # 1. CSV 데이터 로드
    df = pd.read_csv(file_path)  # Pandas를 이용해 CSV 파일 읽기
    print(f"📊 데이터 로드 완료: {df.shape}")  # 데이터 개수 출력

    # 2. 데이터 처리 (평균 값 계산)
    mean_values = df.mean().to_dict()  # 모든 수치형 컬럼의 평균값을 계산

    # ------------------------------
    # 3. MLflow에 데이터 로깅 (컬럼명 정리)
    # ------------------------------
    mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)  # MLflow 서버 주소 설정
    mlflow.set_experiment(EXPERIMENT_NAME)  # 사용할 Experiment 설정

    # 컬럼명 변환 (공백, 괄호 등을 '_'로 변경)
    cleaned_mean_values = {col.replace(" ", "_").replace("(", "").replace(")", ""): val for col, val in mean_values.items()}

    # 새로운 Run을 시작하고 로그 기록
    with mlflow.start_run():
        mlflow.log_params({"source": "MinIO", "file": FILE_NAME})  # 데이터 소스 정보 기록
        mlflow.log_metrics(cleaned_mean_values)  # 변환된 컬럼명으로 메트릭 로깅

    print("✅ MLflow에 데이터 기록 완료!")  # 성공 메시지 출력

# ------------------------------
# 4. DAG 정의
# ------------------------------

# dag의 단위인 각 task를 5분에 한번씩
default_args = {
    "owner": "fisa",  # DAG 소유자
    "start_date": datetime(2025, 3, 19),  # DAG 시작 날짜
    "retries": 1,  # 실패 시 재시도 횟수
    "retry_delay": timedelta(minutes=5),  # 재시도 간격 (5분)
}

# dag자체를 1시간에 한번씩
dag = DAG(
    "00_airflow_mlops",  # DAG 이름
    default_args=default_args,  # 기본 인자 설정
    description="MinIO에서 데이터를 가져와 MLflow에 기록하는 DAG",  # 설명
    schedule_interval="0 * * * *",  # 매 시간 정각에 실행
    catchup=False,  # 과거 실행 안 함
    tags=["minio", "mlflow", "airflow"],  # 태그 설정
)

# ------------------------------
# 5. Task 정의
# ------------------------------

# (1) MinIO에서 데이터를 가져오는 Task
fetch_data_task = PythonOperator(
    task_id="fetch_data",  # Task 이름
    python_callable=fetch_data_from_minio,  # 실행할 함수
    dag=dag,
)

# (2) 데이터 처리 및 MLflow 기록 Task
process_data_task = PythonOperator(
    task_id="process_data",  # Task 이름
    python_callable=process_data,  # 실행할 함수
    provide_context=True,  # XCom을 사용하기 위해 설정
    dag=dag,
)

# ------------------------------
# 6. DAG 실행 순서 정의
# ------------------------------
fetch_data_task >> process_data_task  # fetch_data 실행 후 process_data 실행