운영: 트래픽 분할, 비용, 트러블슈팅

# 트래픽 분할 설정
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}")