엔드포인트 모니터링 (Endpoint Monitoring)

-- system 테이블에서 Percentile 메트릭 계산
SELECT
    endpoint_name,
    DATE(request_time) AS day,
    COUNT(*) AS total_requests,
    ROUND(PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p50_ms,
    ROUND(PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p95_ms,
    ROUND(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p99_ms,
    ROUND(PERCENTILE_CONT(0.999) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p999_ms,
    ROUND(MAX(execution_time_ms), 1) AS max_ms
FROM system.serving.served_model_requests
WHERE DATE(request_time) >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY endpoint_name, DATE(request_time)
ORDER BY day DESC, endpoint_name;

-- 시간대별 평균 RPS(Requests Per Second) 계산
SELECT
    endpoint_name,
    DATE_TRUNC('hour', request_time) AS hour_bucket,
    COUNT(*) AS total_requests,
    -- 1시간(3600초) 기준 평균 RPS
    ROUND(COUNT(*) / 3600.0, 2) AS avg_rps
FROM system.serving.served_model_requests
WHERE DATE(request_time) >= CURRENT_DATE() - INTERVAL 1 DAYS
GROUP BY endpoint_name, DATE_TRUNC('hour', request_time)
ORDER BY hour_bucket DESC;

# 엔드포인트 생성 시 Inference Table 활성화
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import (
    EndpointCoreConfigInput,
    ServedEntityInput,
    AutoCaptureConfigInput,
)

w = WorkspaceClient()

auto_capture = AutoCaptureConfigInput(
    enabled=True,
    catalog_name="ml_prod",
    schema_name="monitoring",
    table_name_prefix="fraud_model"
)

w.serving_endpoints.create(
    name="fraud-detector",
    config=EndpointCoreConfigInput(
        served_entities=[
            ServedEntityInput(
                entity_name="ml_prod.models.fraud_detector",
                entity_version="3",
                workload_size="Small",
                scale_to_zero_enabled=True,
            )
        ],
        auto_capture_config=auto_capture,
    ),
)
# → ml_prod.monitoring.fraud_model_payload 테이블이 자동 생성됩니다

-- Inference Table 스키마 확인
DESCRIBE TABLE ml_prod.monitoring.fraud_model_payload;

-- 주요 컬럼:
-- date              : DATE        (파티션 키)
-- timestamp_ms      : BIGINT      (요청 시각, 밀리초)
-- status_code       : INT         (HTTP 상태 코드)
-- execution_time_ms : DOUBLE      (추론 소요 시간)
-- request           : STRING      (요청 JSON)
-- response          : STRING      (응답 JSON)
-- request_metadata  : MAP<STRING,STRING> (추적 메타데이터)
-- sampling_fraction : DOUBLE      (샘플링 비율)

# 샘플링 비율 설정 예시 (엔드포인트 업데이트 시)
auto_capture = AutoCaptureConfigInput(
    enabled=True,
    catalog_name="ml_prod",
    schema_name="monitoring",
    table_name_prefix="fraud_model",
    # 10% 샘플링: 고트래픽 환경에서 비용 절감
    # 기본값은 100% (sampling_fraction 미설정 시)
)

-- Inference Table에 데이터 보존 정책 설정 (90일)
ALTER TABLE ml_prod.monitoring.fraud_model_payload
SET TBLPROPERTIES (
  'delta.deletedFileRetentionDuration' = 'interval 7 days',
  'delta.logRetentionDuration' = 'interval 90 days'
);

-- 90일 이전 데이터 정기 삭제 (Lakeflow Jobs으로 예약)
DELETE FROM ml_prod.monitoring.fraud_model_payload
WHERE date < CURRENT_DATE() - INTERVAL 90 DAYS;

-- VACUUM으로 물리적 파일 정리
VACUUM ml_prod.monitoring.fraud_model_payload RETAIN 168 HOURS;

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import MonitorInferenceLog, MonitorInferenceLogProblemType

w = WorkspaceClient()

# Inference Table에 Lakehouse Monitor 설정
w.quality_monitors.create(
    table_name="ml_prod.monitoring.fraud_model_payload",
    inference_log=MonitorInferenceLog(
        problem_type=MonitorInferenceLogProblemType.PROBLEM_TYPE_CLASSIFICATION,
        prediction_col="prediction",
        label_col="actual_label",       # Ground Truth (사후 레이블링 필요)
        timestamp_col="timestamp_ms",
        model_id_col="model_version",
        granularities=["1 day", "1 hour"],
    ),
    output_schema_name="ml_prod.monitoring",
)

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import (
    MonitorInferenceLog,
    MonitorInferenceLogProblemType,
    MonitorNotificationsConfig,
    MonitorNotificationsSpec,
)

w = WorkspaceClient()

# 알림이 포함된 모니터 생성
w.quality_monitors.create(
    table_name="ml_prod.monitoring.fraud_model_payload",
    inference_log=MonitorInferenceLog(
        problem_type=MonitorInferenceLogProblemType.PROBLEM_TYPE_CLASSIFICATION,
        prediction_col="prediction",
        label_col="actual_label",
        timestamp_col="timestamp_ms",
        model_id_col="model_version",
        granularities=["1 day"],
    ),
    output_schema_name="ml_prod.monitoring",
    # 드리프트 감지 시 이메일 알림
    notifications=MonitorNotificationsConfig(
        on_new_classification_tag_detected=MonitorNotificationsSpec(
            email_addresses=["ml-team@company.com"]
        )
    ),
    # 모니터 실행 실패 시 알림
    # on_failure=MonitorNotificationsSpec(email_addresses=["ml-team@company.com"])
)

-- 생성된 드리프트 메트릭 조회 (자동 생성 테이블)
SELECT
    window,
    column_name,
    drift_type,
    drift_value,
    threshold_value,
    -- 0.2 초과 시 알림 트리거
    CASE WHEN drift_value > 0.2 THEN 'ALERT' ELSE 'OK' END AS status
FROM ml_prod.monitoring.fraud_model_payload_drift_metrics
WHERE window_start >= CURRENT_DATE() - INTERVAL 7 DAYS
ORDER BY window_start DESC, drift_value DESC;

-- Ground Truth 레이블을 Inference Table에 조인
-- (실제 결과가 나중에 확인되는 경우)
CREATE OR REPLACE TABLE ml_prod.monitoring.fraud_model_labeled AS
SELECT
    p.*,
    g.actual_label
FROM ml_prod.monitoring.fraud_model_payload p
LEFT JOIN ml_prod.gold.fraud_labels g
    ON p.request:customer_id = g.customer_id
    AND p.date = g.event_date;

-- 일별 모델 정확도 추적
SELECT
    date,
    COUNT(*) AS total_predictions,
    SUM(CASE WHEN prediction = actual_label THEN 1 ELSE 0 END) AS correct,
    ROUND(AVG(CASE WHEN prediction = actual_label THEN 1.0 ELSE 0.0 END) * 100, 2) AS accuracy_pct
FROM ml_prod.monitoring.fraud_model_labeled
WHERE actual_label IS NOT NULL
GROUP BY date
ORDER BY date DESC;

# MLflow Tracing은 에이전트 서빙 시 자동 활성화됩니다
# Inference Table에서 트레이스 데이터를 조회할 수 있습니다

import json

# 트레이스 데이터 분석 쿼리 (SQL)
# SELECT
#     request_id,
#     trace:spans[0].name AS root_span,
#     trace:spans[0].attributes.total_tokens AS total_tokens,
#     execution_time_ms
# FROM ml_prod.monitoring.agent_payload
# WHERE date = CURRENT_DATE()
# ORDER BY execution_time_ms DESC
# LIMIT 10;

-- 엔드포인트별 일일 요청량 및 에러율
SELECT
    endpoint_name,
    DATE(request_time) AS day,
    COUNT(*) AS total_requests,
    SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END) AS errors,
    ROUND(
        SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2
    ) AS error_rate_pct,
    ROUND(AVG(execution_time_ms), 1) AS avg_latency_ms,
    ROUND(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p99_latency_ms
FROM system.serving.served_model_requests
WHERE DATE(request_time) >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY endpoint_name, DATE(request_time)
ORDER BY day DESC, endpoint_name;

-- 엔드포인트별 프로비저닝된 동시성 변화 추적
SELECT
    endpoint_name,
    change_time,
    scaled_entity_name,
    previous_scale,
    new_scale
FROM system.serving.endpoint_scaling_events
WHERE DATE(change_time) >= CURRENT_DATE() - INTERVAL 7 DAYS
ORDER BY change_time DESC;

-- LLM 엔드포인트 일별 토큰 사용량 및 추정 비용
SELECT
    DATE(timestamp_ms / 1000) AS day,
    COUNT(*) AS total_requests,
    SUM(request:usage.prompt_tokens) AS total_input_tokens,
    SUM(request:usage.completion_tokens) AS total_output_tokens,
    SUM(request:usage.total_tokens) AS total_tokens,
    -- 대략적인 비용 추정 (모델별로 단가가 다릅니다)
    ROUND(SUM(request:usage.total_tokens) / 1000000.0 * 2.0, 2) AS estimated_cost_usd
FROM ml_prod.monitoring.agent_payload
WHERE date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY DATE(timestamp_ms / 1000)
ORDER BY day DESC;

-- 서빙 엔드포인트의 DBU 사용량 (system.billing 테이블)
SELECT
    usage_date,
    workspace_id,
    sku_name,
    usage_metadata.endpoint_name,
    SUM(usage_quantity) AS total_dbus,
    SUM(usage_quantity * list_price) AS estimated_cost
FROM system.billing.usage
WHERE sku_name LIKE '%SERVING%'
    AND usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY usage_date, workspace_id, sku_name, usage_metadata.endpoint_name
ORDER BY usage_date DESC;

-- 에러율이 5%를 초과하면 알림 트리거
-- (이 쿼리를 Databricks SQL Alert로 등록합니다)
SELECT
    endpoint_name,
    COUNT(*) AS total_requests,
    SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END) AS errors,
    ROUND(
        SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2
    ) AS error_rate_pct
FROM system.serving.served_model_requests
WHERE request_time >= CURRENT_TIMESTAMP() - INTERVAL 1 HOUR
GROUP BY endpoint_name
HAVING error_rate_pct > 5.0;

# Slack Webhook을 통한 알림 예시 (Lakeflow Jobs의 Task로 실행)
import requests
import json

slack_webhook_url = "https://hooks.slack.com/services/T.../B.../..."

def send_alert(endpoint_name, error_rate, p99_latency):
    payload = {
        "text": f":rotating_light: *서빙 엔드포인트 알림*\n"
                f"- 엔드포인트: `{endpoint_name}`\n"
                f"- 에러율: {error_rate}%\n"
                f"- P99 지연시간: {p99_latency}ms"
    }
    requests.post(slack_webhook_url, json=payload)

-- 1. 일별 요청량 및 Percentile 지연 시간 (시계열 차트)
SELECT
    DATE(timestamp_ms / 1000) AS day,
    COUNT(*) AS total_requests,
    ROUND(PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p50_ms,
    ROUND(PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p95_ms,
    ROUND(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY execution_time_ms), 1) AS p99_ms,
    SUM(CASE WHEN status_code != 200 THEN 1 ELSE 0 END) AS error_count,
    ROUND(SUM(CASE WHEN status_code != 200 THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS error_rate_pct
FROM ml_prod.monitoring.fraud_model_payload
WHERE date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY DATE(timestamp_ms / 1000)
ORDER BY day DESC;

-- 2. LLM 엔드포인트: 토큰 사용량 추적 (시계열 차트)
SELECT
    DATE(timestamp_ms / 1000) AS day,
    SUM(request:usage.prompt_tokens) AS total_input_tokens,
    SUM(request:usage.completion_tokens) AS total_output_tokens,
    SUM(request:usage.total_tokens) AS total_tokens,
    ROUND(AVG(execution_time_ms), 0) AS avg_latency
FROM ml_prod.monitoring.agent_payload
WHERE date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY DATE(timestamp_ms / 1000)
ORDER BY day DESC;

-- 3. 에러 분석 (테이블 위젯)
SELECT
    status_code,
    COUNT(*) AS count,
    FIRST(response) AS sample_error
FROM ml_prod.monitoring.fraud_model_payload
WHERE status_code != 200
    AND date = CURRENT_DATE()
GROUP BY status_code;

-- 4. 시간대별 트래픽 패턴 (히트맵)
SELECT
    DAYOFWEEK(timestamp_ms / 1000) AS day_of_week,
    HOUR(timestamp_ms / 1000) AS hour_of_day,
    COUNT(*) AS request_count
FROM ml_prod.monitoring.fraud_model_payload
WHERE date >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY day_of_week, hour_of_day;

-- 5. 엔드포인트별 SLA 준수율 비교 (가로 막대 차트)
-- SLA 기준: P99 < 1000ms, 에러율 < 1%
SELECT
    endpoint_name,
    DATE(request_time) AS day,
    COUNT(*) AS total_requests,
    ROUND(PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY execution_time_ms), 0) AS p99_ms,
    ROUND(SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS error_rate_pct,
    -- SLA 준수 여부
    CASE
        WHEN PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY execution_time_ms) <= 1000
             AND SUM(CASE WHEN status_code >= 400 THEN 1 ELSE 0 END) * 100.0 / COUNT(*) <= 1.0
        THEN 'SLA 준수'
        ELSE 'SLA 위반'
    END AS sla_status
FROM system.serving.served_model_requests
WHERE DATE(request_time) >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY endpoint_name, DATE(request_time)
ORDER BY day DESC, endpoint_name;

-- 6. 비용 추이 (일별, 엔드포인트별)
SELECT
    usage_date,
    usage_metadata.endpoint_name,
    SUM(usage_quantity) AS total_dbus,
    ROUND(SUM(usage_quantity * list_price), 2) AS daily_cost_usd
FROM system.billing.usage
WHERE sku_name LIKE '%SERVING%'
    AND usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY usage_date, usage_metadata.endpoint_name
ORDER BY usage_date DESC;

# 올바른 순서: 엔드포인트 생성과 Inference Table 활성화를 동시에
w.serving_endpoints.create(
    name="my-model",
    config=EndpointCoreConfigInput(
        served_entities=[...],
        auto_capture_config=AutoCaptureConfigInput(enabled=True, ...)  # 처음부터 활성화
    )
)

-- 배포 시점 기준 분포 저장
CREATE TABLE ml_prod.monitoring.fraud_model_baseline AS
SELECT * FROM ml_prod.monitoring.fraud_model_payload
WHERE date BETWEEN '2024-01-01' AND '2024-01-31';  -- 첫 달 데이터

# 클라이언트에서 버전 정보를 메타데이터로 전달
import requests

response = requests.post(
    endpoint_url,
    headers={
        "Authorization": f"Bearer {token}",
        "X-Model-Version": "v3",        # 커스텀 메타데이터
        "X-Client-Id": "recommendation-service"
    },
    json={"inputs": features}
)

-- 일별 비용이 임계값 초과 시 알림 (Databricks SQL Alert로 등록)
SELECT
    usage_date,
    SUM(usage_quantity * list_price) AS daily_cost_usd
FROM system.billing.usage
WHERE sku_name LIKE '%SERVING%'
    AND usage_date = CURRENT_DATE() - INTERVAL 1 DAYS
GROUP BY usage_date
HAVING daily_cost_usd > 100.0;  -- $100 초과 시 알림

-- 잘못된 예시: 전체 테이블 스캔으로 매우 느림
SELECT * FROM ml_prod.monitoring.fraud_model_payload
WHERE timestamp_ms > UNIX_TIMESTAMP(CURRENT_TIMESTAMP() - INTERVAL 1 DAYS) * 1000;

-- 올바른 예시: date 파티션 필터 추가
SELECT * FROM ml_prod.monitoring.fraud_model_payload
WHERE date >= CURRENT_DATE() - INTERVAL 1 DAYS  -- 파티션 프루닝
    AND timestamp_ms > UNIX_TIMESTAMP(CURRENT_TIMESTAMP() - INTERVAL 1 DAYS) * 1000;

# 올바른 예시: 콜드 스타트를 고려한 충분한 타임아웃
response = requests.post(
    endpoint_url,
    json=payload,
    timeout=60  # 60초 타임아웃 (콜드 스타트 포함)
)