개요와 주요 시스템 테이블

system (카탈로그)
├── access           ← 보안 감사 로그
│   ├── audit
│   └── table_lineage (access 스키마의 lineage 뷰)
├── billing          ← 비용 / DBU 사용량
│   ├── usage
│   └── list_prices
├── compute          ← 클러스터 / 웨어하우스
│   ├── clusters
│   └── warehouse_events
├── lakeflow         ← Job / 워크플로우 실행
│   ├── jobs
│   ├── job_tasks
│   └── job_run_timeline
├── lineage          ← 데이터 리니지
│   ├── table_lineage
│   └── column_lineage
└── storage          ← 스토리지 최적화
    └── predictive_optimization_operations_history

# REST API로 특정 스키마 활성화 (예: billing)
curl -X PUT "https://accounts.azuredatabricks.net/api/2.0/accounts/{account_id}/metastores/{metastore_id}/systemschemas/billing" \
  -H "Authorization: Bearer <token>"

-- 지난 30일간 워크로드 유형별 DBU 사용량 및 비용 추정
SELECT
    billing_origin_product,
    sku_name,
    SUM(u.usage_quantity)                              AS total_dbus,
    ROUND(SUM(u.usage_quantity * p.pricing.default), 2) AS estimated_cost_usd
FROM system.billing.usage u
LEFT JOIN system.billing.list_prices p
    ON u.sku_name = p.sku_name
    AND u.usage_date BETWEEN p.price_start_time::DATE AND COALESCE(p.price_end_time::DATE, CURRENT_DATE())
WHERE u.usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY 1, 2
ORDER BY estimated_cost_usd DESC;

-- 일별 비용 추이 (급증 감지용)
SELECT
    usage_date,
    billing_origin_product,
    SUM(usage_quantity) AS daily_dbus
FROM system.billing.usage
WHERE usage_date >= CURRENT_DATE() - INTERVAL 90 DAYS
GROUP BY 1, 2
ORDER BY usage_date, billing_origin_product;

-- 사용자별 DBU 소비 TOP 10
SELECT
    identity_metadata.run_as   AS user_email,
    billing_origin_product     AS product,
    SUM(usage_quantity)        AS total_dbus
FROM system.billing.usage
WHERE usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY 1, 2
ORDER BY total_dbus DESC
LIMIT 10;

-- 최근 7일간 누가 어떤 테이블에 접근했는지 확인
SELECT
    event_date,
    user_identity.email     AS user_email,
    action_name,
    request_params.full_name_arg AS table_name,
    source_ip_address
FROM system.access.audit
WHERE action_name IN ('getTable', 'commandSubmit')
    AND event_date >= CURRENT_DATE() - INTERVAL 7 DAYS
ORDER BY event_date DESC
LIMIT 100;

-- 비정상 접근 패턴 감지: 야간 시간대(22시~06시) 활동
SELECT
    user_identity.email,
    COUNT(*)                    AS late_night_actions,
    COLLECT_SET(action_name)    AS action_types
FROM system.access.audit
WHERE HOUR(event_time) NOT BETWEEN 6 AND 22
    AND event_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY user_identity.email
HAVING COUNT(*) > 10
ORDER BY late_night_actions DESC;

-- 권한 변경 이력 (중요 보안 이벤트)
SELECT
    event_time,
    user_identity.email     AS actor,
    action_name,
    request_params
FROM system.access.audit
WHERE action_name IN (
    'updatePermissions', 'grantPrivileges', 'revokePrivileges',
    'createCredential', 'deleteCredential'
)
    AND event_date >= CURRENT_DATE() - INTERVAL 30 DAYS
ORDER BY event_time DESC;

-- 자동 종료 설정이 없거나 너무 긴 클러스터 (비용 누수 리스크)
SELECT
    cluster_id,
    cluster_name,
    owned_by,
    autotermination_minutes,
    state,
    TIMESTAMPDIFF(HOUR, state_start_time, CURRENT_TIMESTAMP()) AS running_hours
FROM system.compute.clusters
WHERE state = 'RUNNING'
    AND (autotermination_minutes IS NULL OR autotermination_minutes > 120)
ORDER BY running_hours DESC;

-- 클러스터 유형별 분포 파악
SELECT
    cluster_source,
    node_type_id,
    COUNT(DISTINCT cluster_id)  AS cluster_count,
    AVG(autotermination_minutes) AS avg_autotermination_min
FROM system.compute.clusters
WHERE state_start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY 1, 2
ORDER BY cluster_count DESC;

-- 최근 7일간 실패한 Job (에러 메시지 포함)
SELECT
    job_id,
    run_id,
    result_state,
    TIMESTAMPDIFF(MINUTE, start_time, end_time) AS duration_minutes,
    error_message,
    start_time
FROM system.lakeflow.jobs
WHERE result_state = 'FAILED'
    AND start_time >= CURRENT_DATE() - INTERVAL 7 DAYS
ORDER BY start_time DESC
LIMIT 10;

-- Job별 평균 실행 시간 추이 (성능 저하 감지)
SELECT
    job_id,
    DATE(start_time)                                        AS run_date,
    AVG(TIMESTAMPDIFF(MINUTE, start_time, end_time))        AS avg_duration_min,
    COUNT(*)                                                AS run_count,
    SUM(CASE WHEN result_state = 'FAILED' THEN 1 ELSE 0 END) AS fail_count
FROM system.lakeflow.jobs
WHERE start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY job_id, DATE(start_time)
ORDER BY job_id, run_date;

-- Job별 DBU 소비량 (비용 배분 분석)
SELECT
    t.job_id,
    SUM(t.dbu_consumed)                                 AS total_dbu,
    ROUND(SUM(t.dbu_consumed * p.pricing.default), 2)   AS estimated_cost_usd
FROM system.lakeflow.job_run_timeline t
LEFT JOIN system.billing.list_prices p
    ON p.sku_name LIKE '%JOBS%'
    AND t.period_start_time::DATE BETWEEN p.price_start_time::DATE AND COALESCE(p.price_end_time::DATE, CURRENT_DATE())
WHERE t.period_start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY t.job_id
ORDER BY total_dbu DESC
LIMIT 20;

-- 특정 테이블에 의존하는 모든 downstream 테이블
SELECT
    target_table_full_name,
    target_type,
    source_table_full_name
FROM system.lineage.table_lineage
WHERE source_table_full_name = 'production.ecommerce.silver_orders'
ORDER BY target_table_full_name;

-- 특정 컬럼의 upstream 출처 추적
SELECT
    source_table_full_name,
    source_column_name,
    target_table_full_name,
    target_column_name
FROM system.lineage.column_lineage
WHERE target_table_full_name = 'production.ecommerce.gold_revenue'
    AND target_column_name = 'total_amount';

-- 커스텀 태그 기반 팀별 비용 집계
SELECT
    usage_date,
    custom_tags.Team                                        AS team,
    custom_tags.Project                                     AS project,
    billing_origin_product,
    SUM(u.usage_quantity)                                   AS total_dbus,
    ROUND(SUM(u.usage_quantity * p.pricing.default), 2)     AS estimated_cost_usd
FROM system.billing.usage u
LEFT JOIN system.billing.list_prices p
    ON u.sku_name = p.sku_name
    AND u.usage_date BETWEEN p.price_start_time::DATE
        AND COALESCE(p.price_end_time::DATE, CURRENT_DATE())
WHERE u.usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
    AND custom_tags.Team IS NOT NULL
GROUP BY 1, 2, 3, 4
ORDER BY usage_date, team, estimated_cost_usd DESC;

-- 1시간 내 DROP/DELETE 액션 다수 수행한 사용자 탐지
SELECT
    user_identity.email,
    DATE_TRUNC('hour', event_time)  AS hour_bucket,
    COUNT(*)                        AS action_count,
    COLLECT_SET(action_name)        AS actions
FROM system.access.audit
WHERE action_name IN ('deleteTable', 'dropSchema', 'deleteCatalog', 'deleteVolume')
    AND event_date >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY 1, 2
HAVING COUNT(*) >= 5
ORDER BY action_count DESC;

-- 신규 IP에서 처음 접근한 사용자 (기존 접근 IP 목록과 비교)
WITH known_ips AS (
    SELECT user_identity.email, source_ip_address
    FROM system.access.audit
    WHERE event_date BETWEEN CURRENT_DATE() - INTERVAL 60 DAYS
        AND CURRENT_DATE() - INTERVAL 8 DAYS
    GROUP BY 1, 2
),
recent AS (
    SELECT user_identity.email, source_ip_address, MAX(event_time) AS last_seen
    FROM system.access.audit
    WHERE event_date >= CURRENT_DATE() - INTERVAL 7 DAYS
    GROUP BY 1, 2
)
SELECT r.email, r.source_ip_address, r.last_seen
FROM recent r
LEFT JOIN known_ips k
    ON r.email = k.email AND r.source_ip_address = k.source_ip_address
WHERE k.source_ip_address IS NULL;

-- 최근 30일간 billing.usage에 나타나지 않은 클러스터 (미사용 클러스터)
SELECT
    c.cluster_id,
    c.cluster_name,
    c.owned_by,
    c.state,
    c.autotermination_minutes
FROM system.compute.clusters c
LEFT JOIN (
    SELECT DISTINCT usage_metadata.cluster_id AS cluster_id
    FROM system.billing.usage
    WHERE usage_date >= CURRENT_DATE() - INTERVAL 30 DAYS
        AND usage_metadata.cluster_id IS NOT NULL
) active ON c.cluster_id = active.cluster_id
WHERE active.cluster_id IS NULL
    AND c.cluster_source = 'UI'   -- 수동 생성 클러스터만
ORDER BY c.owned_by;

-- 특정 Job이 목표 완료 시각(예: 매일 오전 7시) 이전에 완료되었는지 확인
SELECT
    job_id,
    DATE(start_time)                                            AS run_date,
    MIN(start_time)                                             AS first_start,
    MAX(end_time)                                               AS last_end,
    HOUR(MAX(end_time))                                         AS end_hour,
    TIMESTAMPDIFF(MINUTE, MIN(start_time), MAX(end_time))       AS total_duration_min,
    CASE WHEN HOUR(MAX(end_time)) >= 7 THEN 'SLA_BREACH' ELSE 'OK' END AS sla_status
FROM system.lakeflow.jobs
WHERE job_id IN (123456, 234567)   -- 모니터링 대상 Job ID
    AND result_state = 'SUCCEEDED'
    AND start_time >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY job_id, DATE(start_time)
ORDER BY run_date DESC;

# 매일 전날 비용 집계를 별도 테이블에 적재하는 패턴
spark.sql("""
INSERT INTO main.monitoring.daily_cost_summary
SELECT
    usage_date,
    billing_origin_product,
    custom_tags.Team    AS team,
    SUM(usage_quantity) AS total_dbus
FROM system.billing.usage
WHERE usage_date = CURRENT_DATE() - INTERVAL 1 DAY
GROUP BY 1, 2, 3
""")

Team       = data-platform
Project    = crm-analytics
Env        = production
CostCenter = CC-1234

-- 보안 팀에게 감사 로그 읽기 권한 부여
GRANT SELECT ON TABLE system.access.audit TO `security-team`;

-- 재무 팀에게 빌링 테이블 읽기 권한 부여
GRANT SELECT ON TABLE system.billing.usage TO `finance-team`;
GRANT SELECT ON TABLE system.billing.list_prices TO `finance-team`;