드리프트 감지와 시스템 모니터링

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;