영향도 분석과 거버넌스

-- 1. 테이블 레벨 영향도
SELECT
    target_table_full_name,
    target_type,
    COUNT(*) AS reference_count
FROM system.lineage.table_lineage
WHERE source_table_full_name = 'production.ecommerce.silver_orders'
GROUP BY target_table_full_name, target_type
ORDER BY reference_count DESC;

-- 2. 컬럼 레벨 영향도 (특정 컬럼 변경 시)
SELECT
    target_table_full_name,
    target_column_name,
    source_column_name
FROM system.lineage.column_lineage
WHERE source_table_full_name = 'production.ecommerce.silver_orders'
  AND source_column_name = 'amount'
ORDER BY target_table_full_name;

-- 3. 대시보드 영향도
SELECT DISTINCT
    target_table_full_name AS dashboard_or_query,
    target_type
FROM system.lineage.table_lineage
WHERE source_table_full_name = 'production.ecommerce.silver_orders'
  AND target_type IN ('DASHBOARD', 'QUERY');

-- Feature Table이 변경되면 영향받는 ML 모델 확인
SELECT
    target_table_full_name AS affected_model,
    target_type
FROM system.lineage.table_lineage
WHERE source_table_full_name = 'production.ml.customer_features'
  AND target_type = 'MODEL';
-- → 이 Feature Table로 학습된 모든 모델이 재학습 대상입니다

-- SDP에서 사용하는 소스 테이블이 변경되면 영향받는 Streaming Table
SELECT
    target_table_full_name AS streaming_table,
    target_type
FROM system.lineage.table_lineage
WHERE source_table_full_name = 'production.ecommerce.raw_events'
  AND target_type = 'STREAMING_TABLE';

from databricks.sdk import WorkspaceClient

def check_upstream_quality(table_name):
    """
    특정 테이블의 upstream 품질을 확인하고,
    문제가 있으면 downstream 소유자에게 알림을 보냅니다.
    """
    w = WorkspaceClient()

    # 1. upstream 테이블 확인
    upstream_tables = spark.sql(f"""
        SELECT DISTINCT source_table_full_name
        FROM system.lineage.table_lineage
        WHERE target_table_full_name = '{table_name}'
    """).collect()

    quality_issues = []
    for row in upstream_tables:
        upstream = row.source_table_full_name

        # 2. 각 upstream 테이블의 품질 메트릭 확인
        try:
            quality = spark.sql(f"""
                SELECT *
                FROM {upstream}_profile_metrics
                WHERE log_type = 'DATA_QUALITY'
                  AND quality_check_passed = false
                  AND window_end > DATEADD(hour, -24, CURRENT_TIMESTAMP())
            """).count()

            if quality > 0:
                quality_issues.append(upstream)
        except:
            pass

    if quality_issues:
        # 3. downstream 영향받는 테이블/대시보드 확인
        downstream = spark.sql(f"""
            SELECT DISTINCT target_table_full_name, target_type
            FROM system.lineage.table_lineage
            WHERE source_table_full_name = '{table_name}'
        """).collect()

        print(f"⚠️ 품질 이슈 감지: {quality_issues}")
        print(f"영향받는 downstream: {[r.target_table_full_name for r in downstream]}")
        # Slack/이메일 알림 전송

-- 핵심 Gold 테이블의 upstream이 제시간에 업데이트되었는지 확인
WITH gold_tables AS (
    SELECT 'production.ecommerce.gold_daily_revenue' AS table_name, 6 AS sla_hour
    UNION ALL
    SELECT 'production.ecommerce.gold_customer_360', 8
),
upstream_freshness AS (
    SELECT
        g.table_name AS gold_table,
        g.sla_hour,
        l.source_table_full_name,
        MAX(h.timestamp) AS last_update
    FROM gold_tables g
    JOIN system.lineage.table_lineage l
        ON g.table_name = l.target_table_full_name
    JOIN (
        -- 각 소스 테이블의 마지막 업데이트 시간 (DESCRIBE HISTORY 대용)
        SELECT table_full_name, MAX(event_time) AS timestamp
        FROM system.lineage.table_lineage
        GROUP BY table_full_name
    ) h ON l.source_table_full_name = h.table_full_name
    GROUP BY g.table_name, g.sla_hour, l.source_table_full_name
)
SELECT
    gold_table,
    source_table_full_name,
    last_update,
    CASE
        WHEN last_update < DATEADD(hour, -sla_hour, CURRENT_TIMESTAMP())
        THEN '🔴 SLA 위반'
        ELSE '🟢 정상'
    END AS sla_status
FROM upstream_freshness;

-- PII 태그가 붙은 컬럼의 전파 경로 전체 추적
WITH pii_columns AS (
    SELECT
        catalog_name || '.' || schema_name || '.' || table_name AS table_full_name,
        column_name
    FROM system.information_schema.column_tags
    WHERE tag_name = 'pii' AND tag_value = 'true'
)
SELECT
    p.table_full_name AS pii_source,
    p.column_name AS pii_column,
    cl.target_table_full_name AS propagated_to,
    cl.target_column_name AS target_column
FROM pii_columns p
JOIN system.lineage.column_lineage cl
    ON p.table_full_name = cl.source_table_full_name
    AND p.column_name = cl.source_column_name
ORDER BY pii_source, propagated_to;

-- 분기별 데이터 리니지 감사 보고서
SELECT
    source_table_full_name AS data_source,
    target_table_full_name AS data_destination,
    target_type AS destination_type,
    created_by AS responsible_user,
    MIN(event_time) AS first_seen,
    MAX(event_time) AS last_seen,
    COUNT(*) AS operation_count
FROM system.lineage.table_lineage
WHERE event_time BETWEEN '2025-01-01' AND '2025-03-31'
GROUP BY source_table_full_name, target_table_full_name, target_type, created_by
ORDER BY operation_count DESC;