Model Serving 개요

# Pay-per-token 사용 예시
import mlflow.deployments

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

response = client.predict(
    endpoint="databricks-meta-llama-3-3-70b-instruct",
    inputs={
        "messages": [
            {"role": "system", "content": "당신은 데이터 전문가입니다. 한국어로 답변해 주세요."},
            {"role": "user", "content": "Delta Lake의 ACID 트랜잭션이 왜 중요한가요?"}
        ],
        "max_tokens": 500,
        "temperature": 0.1
    }
)
print(response["choices"][0]["message"]["content"])

# External Model 엔드포인트 생성 (OpenAI 프록시)
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import (
    EndpointCoreConfigInput, ServedEntityInput, ExternalModel, OpenAiConfig
)

w = WorkspaceClient()

w.serving_endpoints.create(
    name="openai-gpt4-proxy",
    config=EndpointCoreConfigInput(
        served_entities=[ServedEntityInput(
            name="gpt4",
            external_model=ExternalModel(
                name="gpt-4",
                provider="openai",
                openai_config=OpenAiConfig(
                    openai_api_key="{{secrets/scope/openai-key}}"
                ),
                task="llm/v1/chat"
            )
        )]
    )
)

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

w = WorkspaceClient()

w.serving_endpoints.create(
    name="fraud-detection-v2",
    config=EndpointCoreConfigInput(
        served_entities=[ServedEntityInput(
            entity_name="catalog.schema.fraud_model",  # UC 등록 모델
            entity_version="3",
            workload_size="Small",                     # Small, Medium, Large
            scale_to_zero_enabled=True,                # 유휴 시 0으로 스케일다운
            workload_type="CPU"                        # CPU 또는 GPU
        )],
        auto_capture_config=AutoCaptureConfigInput(    # Inference Table 자동 활성화
            enabled=True,
            catalog_name="catalog",
            schema_name="schema",
            table_name_prefix="fraud_model"
        )
    )
)

import requests

# REST API로 호출
url = f"https://{workspace_url}/serving-endpoints/fraud-detection-v2/invocations"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# 단건 예측
response = requests.post(url, json={
    "dataframe_records": [
        {"amount": 50000, "merchant_category": "online", "hour": 3, "is_foreign": True}
    ]
}, headers=headers)

print(response.json())  # {"predictions": [0.87]}  (사기 확률 87%)

# Databricks SDK로 호출
w = WorkspaceClient()
response = w.serving_endpoints.query(
    name="fraud-detection-v2",
    dataframe_records=[
        {"amount": 50000, "merchant_category": "online", "hour": 3}
    ]
)

w.serving_endpoints.update_config(
    name="fraud-detection-v2",
    served_entities=[
        ServedEntityInput(
            name="champion",
            entity_name="catalog.schema.fraud_model",
            entity_version="3",
            workload_size="Small",
            scale_to_zero_enabled=True
        ),
        ServedEntityInput(
            name="challenger",
            entity_name="catalog.schema.fraud_model",
            entity_version="4",
            workload_size="Small",
            scale_to_zero_enabled=True
        )
    ],
    traffic_config={
        "routes": [
            {"served_model_name": "champion", "traffic_percentage": 80},
            {"served_model_name": "challenger", "traffic_percentage": 20}
        ]
    }
)

-- Inference Table에서 모니터링 쿼리
SELECT
    DATE(timestamp) AS day,
    COUNT(*) AS total_requests,
    AVG(latency_ms) AS avg_latency,
    SUM(CASE WHEN status_code != 200 THEN 1 ELSE 0 END) AS error_count
FROM catalog.schema.fraud_model_inference_log
GROUP BY DATE(timestamp)
ORDER BY day DESC;