Skip to main content

​
왜 커스텀 모델 배포가 필요한가?

Databricks Model Serving은 등록된 MLflow 모델을 REST API 엔드포인트로 배포하는 관리형 서비스입니다. Foundation Model API (FMAPI) 가 제공하는 사전 학습 모델(llama, mixtral 등)로 충분하지 않은 경우, 커스텀 모델 배포가 필요합니다.

​
Foundation Model API vs. 커스텀 모델 배포

구분Foundation Model API커스텀 모델 배포
모델 제어Databricks가 제공하는 모델만 사용직접 학습/파인튜닝한 모델 사용 가능
전처리/후처리불가 (입출력 형식 고정)완전한 자유도
비용토큰당 종량제 (Pay-per-token)인스턴스 시간당 과금
운영 복잡도낮음 (인프라 자동 관리)높음 (의존성, 스케일링 직접 설정)
레이턴시공유 인프라 (변동 가능)전용 인스턴스 (예측 가능)
규정 준수데이터가 외부 모델 통과워크스페이스 내부에서 처리

​
커스텀 배포가 필요한 경우

사용 사례설명
전후처리 파이프라인입력 데이터 정규화, 피처 엔지니어링, 출력 후처리가 필요한 경우
파인튜닝 모델도메인 특화 데이터로 Fine-tuning한 모델을 배포해야 하는 경우
자체 학습 모델사내 데이터로 처음부터 학습한 ML/DL 모델
앙상블 모델여러 모델의 예측을 결합해야 하는 경우
외부 라이브러리 의존표준 MLflow flavor에 없는 프레임워크를 사용하는 경우
비즈니스 로직 포함예측값에 규칙 기반 로직을 적용해야 하는 경우
GPU 추론대규모 딥러닝 모델을 GPU로 서빙해야 하는 경우
데이터 거버넌스입력 데이터가 외부 서비스로 나가면 안 되는 경우
모델 서빙 엔드포인트 (Model Serving Endpoint) 란 학습된 ML 모델을 REST API로 노출하여, 애플리케이션에서 실시간으로 예측을 요청할 수 있게 해주는 서비스입니다. Databricks에서는 인프라 관리, 오토스케일링, 모니터링을 자동으로 처리합니다.

​
배포 가능한 모델 유형

배포 유형설명포함 모델
표준 Flavor기본 제공 MLflow Flavor를 사용합니다sklearn, xgboost, lightgbm, spark
커스텀 PyFuncmlflow.pyfunc.PythonModel을 상속하여 자유로운 추론 로직을 구현합니다전처리/후처리, 멀티모델 앙상블 등
외부 모델외부 LLM 제공사를 프록시합니다OpenAI, Anthropic, Cohere
모델 유형특징적합한 경우
표준 MLflow Flavormlflow.sklearn, mlflow.xgboost 등으로 저장한 모델단순 predict 호출로 충분한 경우
커스텀 PyFuncmlflow.pyfunc.PythonModel을 상속하여 자유로운 추론 로직 구현전후처리, 앙상블, 비즈니스 로직이 필요한 경우
외부 모델 (External Models)OpenAI, Anthropic 등 외부 API를 Databricks 엔드포인트로 프록시외부 LLM API에 거버넌스/레이트 리밋을 적용할 때

​
모델 시그니처 정의 (Model Signature)

모델 시그니처 (Model Signature) 는 모델의 입력/출력 스키마를 명시적으로 정의합니다. 시그니처가 있으면 서빙 시 입력 데이터 유효성 검사가 자동으로 수행되어 오류를 조기에 감지할 수 있습니다. 
import mlflow
from mlflow.models import ModelSignature, infer_signature
from mlflow.types.schema import Schema, ColSpec, ParamSchema, ParamSpec

# 방법 1: 학습 데이터에서 자동 추론 (권장)
import pandas as pd
import numpy as np

X_sample = pd.DataFrame({
    "amount": [50000.0, 25.0],
    "merchant_category": ["online_retail", "grocery"],
    "hour": [3, 12],
    "is_international": [1, 0]
})
y_sample = pd.DataFrame({
    "fraud_probability": [0.85, 0.02],
    "is_fraud": [1, 0]
})

# 입력/출력 데이터로 시그니처 자동 추론
signature = infer_signature(X_sample, y_sample)
print(signature)
# inputs: [amount: double, merchant_category: string, hour: long, is_international: long]
# outputs: [fraud_probability: double, is_fraud: long]

# 방법 2: 수동으로 시그니처 정의 (정밀한 제어가 필요할 때)
signature = ModelSignature(
    inputs=Schema([
        ColSpec("double", "amount"),
        ColSpec("string", "merchant_category"),
        ColSpec("long", "hour"),
        ColSpec("integer", "is_international")
    ]),
    outputs=Schema([
        ColSpec("double", "fraud_probability"),
        ColSpec("integer", "is_fraud"),
        ColSpec("string", "risk_level")
    ]),
    # params: 추론 시 동적으로 변경할 수 있는 파라미터 (선택)
    params=ParamSchema([
        ParamSpec("threshold", "double", 0.7)
    ])
)

# 방법 3: params를 활용한 동적 추론 파라미터
class FraudModelWithParams(mlflow.pyfunc.PythonModel):
    def load_context(self, context):
        import joblib
        self.model = joblib.load(context.artifacts["model_path"])
        self.scaler = joblib.load(context.artifacts["scaler_path"])

    def predict(self, context, model_input, params=None):
        # params에서 임계값을 런타임에 받음 (기본값 0.7)
        threshold = (params or {}).get("threshold", 0.7)

        scaled_input = self.scaler.transform(model_input)
        probabilities = self.model.predict_proba(scaled_input)[:, 1]

        return pd.DataFrame({
            "fraud_probability": probabilities,
            "is_fraud": (probabilities >= threshold).astype(int)
        })

# params가 포함된 시그니처 생성
X_sample = X_test[:5]
predictions = FraudModelWithParams().predict(None, X_sample)
signature = infer_signature(
    X_sample,
    predictions,
    params={"threshold": 0.7}   # 기본 파라미터 값
)
시그니처 없이 배포하면 입력 형식 오류가 서빙 환경에서 런타임에 발생합니다. input_example과 함께 signature를 항상 명시하는 것을 권장합니다.

​
PyFunc 커스텀 모델 작성법

mlflow.pyfunc.PythonModel을 상속하면 load_context()와 predict() 메서드를 직접 구현하여 자유로운 추론 로직을 만들 수 있습니다.

​
기본 구조

import mlflow.pyfunc
import pandas as pd
import numpy as np

class FraudDetectionModel(mlflow.pyfunc.PythonModel):
    """전처리 + 모델 추론 + 후처리를 하나로 묶는 커스텀 모델"""

    def load_context(self, context):
        """모델 로드 시 한 번 실행 (무거운 초기화 작업)"""
        import joblib

        # artifacts에서 모델과 전처리기 로드
        self.model = joblib.load(context.artifacts["model_path"])
        self.scaler = joblib.load(context.artifacts["scaler_path"])
        self.threshold = 0.7  # 커스텀 임계값

    def predict(self, context, model_input, params=None):
        """추론 요청마다 실행되는 메서드"""
        # 1. 전처리
        scaled_input = self.scaler.transform(model_input)

        # 2. 예측 확률 계산
        probabilities = self.model.predict_proba(scaled_input)[:, 1]

        # 3. 후처리 (커스텀 임계값 + 리스크 등급)
        results = pd.DataFrame({
            "fraud_probability": probabilities,
            "is_fraud": (probabilities >= self.threshold).astype(int),
            "risk_level": pd.cut(
                probabilities,
                bins=[0, 0.3, 0.7, 1.0],
                labels=["LOW", "MEDIUM", "HIGH"]
            )
        })

        return results

​
커스텀 모델 저장 및 등록

import joblib
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler

# 모델과 스케일러 학습
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
model = GradientBoostingClassifier(n_estimators=200)
model.fit(X_train_scaled, y_train)

# 아티팩트 저장
joblib.dump(model, "/tmp/model.joblib")
joblib.dump(scaler, "/tmp/scaler.joblib")

# 시그니처 자동 추론
sample_input = pd.DataFrame(X_test[:3], columns=feature_cols)
sample_output = FraudDetectionModel().predict(None, sample_input)
signature = infer_signature(sample_input, sample_output)

# 커스텀 모델을 MLflow에 로깅
with mlflow.start_run(run_name="fraud-custom-pyfunc"):
    model_info = mlflow.pyfunc.log_model(
        artifact_path="fraud_model",
        python_model=FraudDetectionModel(),
        artifacts={
            "model_path": "/tmp/model.joblib",
            "scaler_path": "/tmp/scaler.joblib"
        },
        conda_env={
            "dependencies": [
                "python=3.10",
                {"pip": ["scikit-learn==1.4.0", "pandas>=2.0", "numpy>=1.24"]}
            ]
        },
        signature=signature,
        input_example=sample_input,
        registered_model_name="catalog.schema.fraud_detection_custom"
    )
의존성 관리 주의: conda_env 또는 pip_requirements를 명시하지 않으면, 로컬 환경의 패키지 버전이 서빙 환경과 달라 오류가 발생할 수 있습니다. 반드시 학습에 사용한 주요 라이브러리 버전을 명시하십시오.