모니터링과 알림

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# 최근 실행 이력 조회
runs = w.jobs.list_runs(
    job_id=12345,
    limit=10,
    expand_tasks=True
)

for run in runs:
    print(f"Run {run.run_id}: {run.state.result_state} "
          f"({run.execution_duration / 1000:.0f}초)")

    # 실패한 태스크 확인
    if run.state.result_state == "FAILED":
        for task in run.tasks:
            if task.state.result_state == "FAILED":
                print(f"  실패 태스크: {task.task_key}")
                print(f"  에러: {task.state.state_message}")

# 알림 목적지 설정 후 Job에 연동
webhook_notifications:
  on_start:
    - id: "${var.slack_info_webhook}"
  on_success:
    - id: "${var.slack_info_webhook}"
  on_failure:
    - id: "${var.slack_alert_webhook}"
    - id: "${var.pagerduty_webhook}"
  on_duration_warning_threshold_exceeded:
    - id: "${var.slack_alert_webhook}"

email_notifications:
  on_failure:
    - "data-oncall@company.com"
    - "team-lead@company.com"
  on_duration_warning_threshold_exceeded:
    - "data-oncall@company.com"

-- 최근 7일간 Job 실행 현황 요약
SELECT
    job_id,
    job_name,
    COUNT(*) AS total_runs,
    SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) AS success_count,
    SUM(CASE WHEN result_state = 'FAILED' THEN 1 ELSE 0 END) AS failure_count,
    ROUND(
        SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1
    ) AS success_rate_pct,
    AVG(TIMESTAMPDIFF(MINUTE, period_start_time, period_end_time)) AS avg_duration_min,
    MAX(TIMESTAMPDIFF(MINUTE, period_start_time, period_end_time)) AS max_duration_min
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY job_id, job_name
ORDER BY failure_count DESC;

-- 자주 실패하는 태스크 Top 10
SELECT
    t.job_id,
    j.job_name,
    t.task_key,
    COUNT(*) AS failure_count,
    COLLECT_SET(t.result_state) AS failure_types,
    MAX(t.period_start_time) AS last_failure_time
FROM system.lakeflow.job_task_run_timeline t
JOIN system.lakeflow.job_run_timeline j
    ON t.run_id = j.run_id
WHERE t.result_state IN ('FAILED', 'TIMEDOUT')
    AND t.period_start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY t.job_id, j.job_name, t.task_key
ORDER BY failure_count DESC
LIMIT 10;

-- 최근 30일 Job별 DBU 사용량 (비용 상위 20)
SELECT
    usage_metadata.job_id,
    j.job_name,
    SUM(usage_quantity) AS total_dbus,
    COUNT(DISTINCT usage_date) AS active_days,
    ROUND(SUM(usage_quantity) / COUNT(DISTINCT usage_date), 1) AS avg_daily_dbus,
    SUM(usage_quantity * list_price) AS estimated_cost_usd
FROM system.billing.usage u
LEFT JOIN system.lakeflow.job_run_timeline j
    ON u.usage_metadata.job_id = j.job_id
WHERE u.usage_metadata.job_id IS NOT NULL
    AND u.usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY usage_metadata.job_id, j.job_name
ORDER BY total_dbus DESC
LIMIT 20;

-- 주간 비용 추이 (Job 컴퓨트)
SELECT
    DATE_TRUNC('WEEK', usage_date) AS week,
    sku_name,
    SUM(usage_quantity) AS total_dbus,
    SUM(usage_quantity * list_price) AS estimated_cost_usd
FROM system.billing.usage
WHERE usage_metadata.job_id IS NOT NULL
    AND usage_date >= CURRENT_DATE() - INTERVAL 90 DAYS
GROUP BY DATE_TRUNC('WEEK', usage_date), sku_name
ORDER BY week DESC, total_dbus DESC;

-- Spot 인스턴스 미사용 Job 탐지 (비용 절감 기회)
SELECT DISTINCT
    usage_metadata.job_id,
    sku_name,
    SUM(usage_quantity) AS total_dbus
FROM system.billing.usage
WHERE usage_metadata.job_id IS NOT NULL
    AND sku_name LIKE '%ALL_PURPOSE%'   -- Job Cluster가 아닌 All-Purpose 사용
    AND usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY usage_metadata.job_id, sku_name
HAVING SUM(usage_quantity) > 100
ORDER BY total_dbus DESC;

-- 오늘의 Job 실행 현황 (대시보드 카드용)
SELECT
    COUNT(*) AS total_runs,
    SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) AS succeeded,
    SUM(CASE WHEN result_state = 'FAILED' THEN 1 ELSE 0 END) AS failed,
    SUM(CASE WHEN result_state IS NULL THEN 1 ELSE 0 END) AS running
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= CURRENT_DATE();

-- 평균 대비 2배 이상 오래 걸린 실행 (최근 24시간)
WITH job_stats AS (
    SELECT
        job_id,
        job_name,
        AVG(TIMESTAMPDIFF(MINUTE, period_start_time, period_end_time)) AS avg_duration,
        STDDEV(TIMESTAMPDIFF(MINUTE, period_start_time, period_end_time)) AS stddev_duration
    FROM system.lakeflow.job_run_timeline
    WHERE period_start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
        AND result_state = 'SUCCESS'
    GROUP BY job_id, job_name
)
SELECT
    r.job_id,
    r.job_name,
    r.run_id,
    TIMESTAMPDIFF(MINUTE, r.period_start_time, r.period_end_time) AS actual_duration_min,
    ROUND(s.avg_duration, 1) AS avg_duration_min,
    ROUND(
        TIMESTAMPDIFF(MINUTE, r.period_start_time, r.period_end_time) / s.avg_duration, 1
    ) AS duration_ratio
FROM system.lakeflow.job_run_timeline r
JOIN job_stats s ON r.job_id = s.job_id
WHERE r.period_start_time >= CURRENT_DATE() - INTERVAL 1 DAY
    AND TIMESTAMPDIFF(MINUTE, r.period_start_time, r.period_end_time) > s.avg_duration * 2
ORDER BY duration_ratio DESC;

-- 최근 30일 일별 성공률 (라인 차트용)
SELECT
    DATE(period_start_time) AS run_date,
    COUNT(*) AS total_runs,
    ROUND(
        SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1
    ) AS success_rate_pct
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
    AND result_state IS NOT NULL
GROUP BY DATE(period_start_time)
ORDER BY run_date;