시스템 테이블 활용 시나리오와 베스트 프랙티스

-- 최근 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`;