[MlOps] Airflow 학습 DAG부터 Ray Serve 추론, Minikube까지

학습 DAG 만들기 

Airflow와 MLflow가 준비됐다면 이제 학습 Pipeline을 하나 만들어보겠습니다.
이 DAG은 Airflow가 실행하는 단일 Python Task로,
MLflow에 학습 Log를 남기고 임의의 정확도를 기록합니다.

 

~/hybrid-mlops-demo/airflow/dags/train_pipeline.py

cat > ~/hybrid-mlops-demo/airflow/dags/train_pipeline.py <<'PY'
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import mlflow, os, time, random

def train_model():
    mlflow.set_tracking_uri("http://host.docker.internal:5000")
    mlflow.set_experiment("hybrid-demo")
    with mlflow.start_run(run_name="mock-train"):
        time.sleep(1)
        acc = round(0.90 + random.random()*0.05, 4)
        mlflow.log_param("epochs", 3)
        mlflow.log_metric("accuracy", acc)
        mlflow.log_artifact(__file__)

with DAG(
    "train_pipeline",
    start_date=datetime(2025,1,1),
    schedule_interval=None,
    catchup=False,
    tags=["demo","mlflow"],
) as dag:
    PythonOperator(task_id="train", python_callable=train_model)
PY

 

 

 

실행 방법

Airflow Web UI 접속 http://localhost:8080 (ID/PW : airflow / airflow)

train_pipeline DAG 활성화 → Trigger 클릭

 

 

 

---

“클라우드 추론” 시뮬레이터 — Ray Serve

이번에는 추론 API Server를 흉내 내볼 차례입니다.
GPU가 없어도 동작되고 나중에 AWS EKS 나 Kubeflow 에 배포될 Ray Serve Model Server의 원형이 될 예정입니다. 

 

cat > ~/hybrid-mlops-demo/ray_inference/serve_app.py <<'PY'
from ray import serve
from fastapi import FastAPI
import time

app = FastAPI()
serve.start(detached=True)

@serve.deployment(route_prefix="/inference")
@serve.ingress(app)
class InferenceService:
    def __init__(self):
        # 여기서 실제론 모델 로드. 데모에선 0.3초 지연으로 대체
        time.sleep(0.3)

    @app.post("/")
    def infer(self, payload: dict):
        time.sleep(0.5)  # 추론 지연 시뮬
        return {"prediction": "vehicle", "confidence": 0.973}

InferenceService.deploy()
PY

 

 

실행: 

pip install --user "ray[serve]" fastapi uvicorn
python ~/hybrid-mlops-demo/ray_inference/serve_app.py
# 새 터미널에서 호출 테스트
curl -s -X POST http://127.0.0.1:8000/inference -H 'Content-Type: application/json' -d '{"image":"frame001.jpg"}' | jq .

실행 및 테스트

1. 필요한 패키지 설치

pip install --user "ray[serve]" fastapi uvicorn

1. 서버 실행

python ~/hybrid-mlops-demo/ray_inference/serve_app.py

1. 새 터미널에서 테스트 요청

curl -s -X POST <http://127.0.0.1:8000/inference> \\
  -H 'Content-Type: application/json' \\
  -d '{"image":"frame001.jpg"}' | jq .

결과 예시

{
  "prediction": "vehicle",
  "confidence": 0.973
}

이 서버는 로컬에서도 GPU 없이 동작하며,

차후 EKS나 Ray Cluster에 올릴 때 바로 재사용할 예정입니다.

 

 

 

 

 

---

Minikube 유틸(Optional)

클러스터 시작/중지 스크립트

Local Kubernetes Cluster를 쉽게 켜고 끄기 위한 Script를 만듭니다,

~/hybrid-mlops-demo/infra/minikube/start_minikube.sh

cat > ~/hybrid-mlops-demo/infra/minikube/start_minikube.sh <<'SH'
#!/usr/bin/env bash
set -euo pipefail
minikube start --driver=docker --cpus=4 --memory=6144
kubectl get nodes
SH
chmod +x ~/hybrid-mlops-demo/infra/minikube/start_minikube.sh

 

사용법 

~/hybrid-mlops-demo/infra/minikube/start_minikube.sh

 

 

 

정리 !

Airflow DAG	학습 Pipeline 실행	Airflow UI → Trigger
MLflow	학습 결과 Tracking	http://localhost:5000
Ray Serve	추론 API Simulation	python serve_app.py → curl 요청
Minikube	Local K8s Cluster	start_minikube.sh

 

 

다음 단계

이제 로컬에서 학습 → 추적 → 추론 의 전체 플로우가 완성되었습니다.
다음 편에서는 이 파이프라인을 Argo CD + EKS 환경으로 확장하여
클라우드 네이티브 MLOps로 전환하는 과정을 다뤄보겠습니다.

 

 

https://github.com/daeun-ops/hybrid-mlops-demo