커스텀 모델 배포 (Custom Model Deployment)

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}   # 기본 파라미터 값
)

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"
    )

class EnsembleModel(mlflow.pyfunc.PythonModel):
    """XGBoost + LightGBM + Logistic Regression 앙상블"""

    def load_context(self, context):
        import joblib
        self.models = {
            "xgb": joblib.load(context.artifacts["xgb_model"]),
            "lgbm": joblib.load(context.artifacts["lgbm_model"]),
            "lr": joblib.load(context.artifacts["lr_model"]),
        }
        # 각 모델의 가중치 (검증 세트 성능 기반)
        self.weights = {"xgb": 0.5, "lgbm": 0.35, "lr": 0.15}

    def predict(self, context, model_input, params=None):
        # 각 모델로 예측 후 가중 평균
        weighted_pred = sum(
            self.weights[name] * model.predict_proba(model_input)[:, 1]
            for name, model in self.models.items()
        )

        return pd.DataFrame({
            "ensemble_probability": weighted_pred,
            "prediction": (weighted_pred >= 0.5).astype(int)
        })

class FineTunedEmbeddingModel(mlflow.pyfunc.PythonModel):
    """파인튜닝된 sentence-transformers 모델"""

    def load_context(self, context):
        from sentence_transformers import SentenceTransformer

        # 파인튜닝된 모델 경로에서 로드
        model_dir = context.artifacts["finetuned_model_dir"]
        self.model = SentenceTransformer(model_dir)
        self.model.eval()

    def predict(self, context, model_input, params=None):
        # 입력이 DataFrame인 경우 텍스트 컬럼 추출
        if isinstance(model_input, pd.DataFrame):
            texts = model_input["text"].tolist()
        else:
            texts = list(model_input)

        # 임베딩 생성 (배치 처리)
        batch_size = (params or {}).get("batch_size", 32)
        embeddings = self.model.encode(
            texts,
            batch_size=batch_size,
            normalize_embeddings=True
        )

        return pd.DataFrame(
            embeddings,
            columns=[f"dim_{i}" for i in range(embeddings.shape[1])]
        )

class CreditScoringModel(mlflow.pyfunc.PythonModel):
    """ML 예측 + 하드 룰(Hard Rule)을 결합한 신용 평가 모델"""

    def load_context(self, context):
        import joblib
        self.ml_model = joblib.load(context.artifacts["ml_model"])

    def _apply_hard_rules(self, df, ml_score):
        """비즈니스 룰로 ML 점수를 오버라이드"""
        final_score = ml_score.copy()

        # 하드 룰: 연체 이력이 있으면 무조건 거절
        final_score[df["delinquency_count"] > 0] = 0.0

        # 하드 룰: 신용 기간 1년 미만은 최대 0.4로 제한
        new_customers = df["credit_age_months"] < 12
        final_score[new_customers] = final_score[new_customers].clip(upper=0.4)

        return final_score

    def predict(self, context, model_input, params=None):
        ml_score = self.ml_model.predict_proba(model_input)[:, 1]
        final_score = self._apply_hard_rules(model_input, ml_score)

        return pd.DataFrame({
            "ml_score": ml_score,
            "final_score": final_score,
            "approved": (final_score >= 0.5).astype(int)
        })

import mlflow.deployments

client = mlflow.deployments.get_deploy_client("databricks")

# 엔드포인트 생성
endpoint = client.create_endpoint(
    name="fraud-detection-v2",
    config={
        "served_entities": [{
            "entity_name": "catalog.schema.fraud_detection_custom",
            "entity_version": "3",
            "workload_size": "Small",           # Small / Medium / Large
            "scale_to_zero_enabled": True,      # 트래픽 없을 때 0으로 축소
            "workload_type": "CPU"              # CPU 또는 GPU_SMALL / GPU_MEDIUM / GPU_LARGE
        }],
        "auto_capture_config": {
            "catalog_name": "catalog",
            "schema_name": "schema",
            "table_name_prefix": "fraud_endpoint"
        }
    }
)

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import (
    EndpointCoreConfigInput,
    ServedEntityInput,
    AutoCaptureConfigInput
)

w = WorkspaceClient()

# 엔드포인트 생성
endpoint = w.serving_endpoints.create(
    name="fraud-detection-v2",
    config=EndpointCoreConfigInput(
        served_entities=[
            ServedEntityInput(
                entity_name="catalog.schema.fraud_detection_custom",
                entity_version="3",
                workload_size="Small",
                scale_to_zero_enabled=True,
                workload_type="CPU"
            )
        ],
        auto_capture_config=AutoCaptureConfigInput(
            catalog_name="catalog",
            schema_name="schema",
            table_name_prefix="fraud_endpoint",
            enabled=True
        )
    )
)

# 엔드포인트 준비 대기
endpoint = w.serving_endpoints.wait_get_serving_endpoint_not_updating(
    name="fraud-detection-v2"
)
print(f"Endpoint state: {endpoint.state.ready}")

curl -X POST "https://<workspace-url>/api/2.0/serving-endpoints" \
  -H "Authorization: Bearer $DATABRICKS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "fraud-detection-v2",
    "config": {
      "served_entities": [{
        "entity_name": "catalog.schema.fraud_detection_custom",
        "entity_version": "3",
        "workload_size": "Small",
        "scale_to_zero_enabled": true
      }]
    }
  }'

# GPU 엔드포인트 생성 예시
endpoint = client.create_endpoint(
    name="embedding-model-gpu",
    config={
        "served_entities": [{
            "entity_name": "catalog.schema.sentence_transformer",
            "entity_version": "1",
            "workload_size": "Small",
            "workload_type": "GPU_SMALL",       # GPU 타입 지정
            "scale_to_zero_enabled": True
        }]
    }
)

# 트래픽 분할 설정
client = mlflow.deployments.get_deploy_client("databricks")

client.update_endpoint(
    endpoint="fraud-detection-v2",
    config={
        "served_entities": [
            {
                "entity_name": "catalog.schema.fraud_detection_custom",
                "entity_version": "3",
                "workload_size": "Small",
                "scale_to_zero_enabled": False
            },
            {
                "entity_name": "catalog.schema.fraud_detection_custom",
                "entity_version": "4",
                "workload_size": "Small",
                "scale_to_zero_enabled": False
            }
        ],
        "traffic_config": {
            "routes": [
                {"served_model_name": "fraud_detection_custom-3", "traffic_percentage": 80},
                {"served_model_name": "fraud_detection_custom-4", "traffic_percentage": 20}
            ]
        }
    }
)

# Inference Table 활성화 (엔드포인트 생성 또는 업데이트 시)
config = {
    "served_entities": [...],
    "auto_capture_config": {
        "catalog_name": "main",
        "schema_name": "serving_logs",
        "table_name_prefix": "fraud_endpoint",   # 테이블명: fraud_endpoint_payload
        "enabled": True
    }
}

-- Inference Table 쿼리 예시
-- 테이블명: {catalog}.{schema}.{prefix}_payload
SELECT
    timestamp,
    request_id,
    databricks_request_id,
    -- 요청 데이터 파싱
    request:dataframe_records[0].amount::double AS amount,
    request:dataframe_records[0].merchant_category::string AS merchant_category,
    -- 응답 데이터 파싱
    response:predictions[0].fraud_probability::double AS fraud_probability,
    response:predictions[0].is_fraud::int AS is_fraud,
    -- 메타데이터
    status_code,
    execution_duration_ms
FROM main.serving_logs.fraud_endpoint_payload
WHERE timestamp >= current_date() - interval 7 days
ORDER BY timestamp DESC;

# Python에서 Inference Table 조회
from databricks.connect import DatabricksSession

spark = DatabricksSession.builder.getOrCreate()

# 최근 1시간 요청 분석
df = spark.sql("""
    SELECT
        date_trunc('minute', timestamp) AS minute,
        count(*) AS request_count,
        avg(execution_duration_ms) AS avg_latency_ms,
        percentile_approx(execution_duration_ms, 0.95) AS p95_latency_ms,
        sum(CASE WHEN status_code != 200 THEN 1 ELSE 0 END) AS error_count
    FROM main.serving_logs.fraud_endpoint_payload
    WHERE timestamp >= now() - interval 1 hour
    GROUP BY 1
    ORDER BY 1
""")
df.display()

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# 엔드포인트 상태 조회
endpoint = w.serving_endpoints.get(name="fraud-detection-v2")
print(f"State: {endpoint.state.ready}")
print(f"Config update: {endpoint.state.config_update}")

# 이벤트 로그 조회 (디버깅용)
for event in w.serving_endpoints.list_serving_endpoint_events(name="fraud-detection-v2"):
    print(f"[{event.timestamp}] {event.type}: {event.message}")

# Provisioned Concurrency 설정 (최소 인스턴스 보장)
endpoint = client.create_endpoint(
    name="fraud-detection-prod",
    config={
        "served_entities": [{
            "entity_name": "catalog.schema.fraud_detection_custom",
            "entity_version": "3",
            "workload_size": "Small",
            "scale_to_zero_enabled": False,     # 항상 최소 1개 인스턴스 유지
            "workload_type": "CPU"
        }]
    }
)

-- Inference Table 데이터 보존 기간 설정 (90일 후 자동 삭제)
ALTER TABLE main.serving_logs.fraud_endpoint_payload
SET TBLPROPERTIES (
    'delta.deletedFileRetentionDuration' = 'interval 90 days'
);

# 비용 절감: 개발 환경은 scale_to_zero 활성화, 프로덕션은 비활성화
def create_endpoint_by_env(env: str, model_name: str, version: str):
    is_prod = env == "production"
    return client.create_endpoint(
        name=f"fraud-detection-{env}",
        config={
            "served_entities": [{
                "entity_name": model_name,
                "entity_version": version,
                "workload_size": "Medium" if is_prod else "Small",
                "scale_to_zero_enabled": not is_prod,
                "workload_type": "CPU"
            }]
        }
    )

import requests
import json

# 방법 1: requests 라이브러리
workspace_url = "https://<workspace-url>"
token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()

url = f"{workspace_url}/serving-endpoints/fraud-detection-v2/invocations"
headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json"
}

# DataFrame 레코드 형식
payload = {
    "dataframe_records": [
        {"amount": 50000, "merchant_category": "online_retail", "hour": 3, "is_international": 1},
        {"amount": 25, "merchant_category": "grocery", "hour": 12, "is_international": 0}
    ]
}

response = requests.post(url, json=payload, headers=headers)
predictions = response.json()
print(json.dumps(predictions, indent=2))

# 방법 2: Databricks SDK
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

response = w.serving_endpoints.query(
    name="fraud-detection-v2",
    dataframe_records=[
        {"amount": 50000, "merchant_category": "online_retail", "hour": 3, "is_international": 1}
    ]
)
print(response.predictions)

# 방법 3: MLflow Deployments Client
client = mlflow.deployments.get_deploy_client("databricks")

response = client.predict(
    endpoint="fraud-detection-v2",
    inputs={
        "dataframe_records": [
            {"amount": 50000, "merchant_category": "online_retail", "hour": 3}
        ]
    }
)
print(response)

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# 엔드포인트 상태 조회
endpoint = w.serving_endpoints.get(name="fraud-detection-v2")
print(f"State: {endpoint.state.ready}")
print(f"Config update: {endpoint.state.config_update}")

# 이벤트 로그 조회 (디버깅용)
for event in w.serving_endpoints.list_serving_endpoint_events(name="fraud-detection-v2"):
    print(f"[{event.timestamp}] {event.type}: {event.message}")