SDP 이벤트 로그 실전 패턴

-- 파이프라인 일별 실행 시간 추이 (점점 느려지고 있는지 확인)
WITH daily_runs AS (
    SELECT
        DATE(timestamp) AS run_date,
        origin.update_id,
        MIN(timestamp) AS start_time,
        MAX(timestamp) AS end_time
    FROM event_log("pipeline-id-here")
    WHERE event_type = 'user_action'
    GROUP BY 1, 2
)
SELECT
    run_date,
    COUNT(DISTINCT update_id) AS num_runs,
    ROUND(AVG(TIMESTAMPDIFF(MINUTE, start_time, end_time)), 1) AS avg_duration_min,
    MAX(TIMESTAMPDIFF(MINUTE, start_time, end_time)) AS max_duration_min
FROM daily_runs
WHERE start_time IS NOT NULL AND end_time IS NOT NULL
GROUP BY run_date
ORDER BY run_date DESC
LIMIT 30;

-- 주간 데이터 품질 리포트
WITH weekly_quality AS (
    SELECT
        DATE_TRUNC('WEEK', timestamp) AS week_start,
        details:flow_progress:flow_name AS flow_name,
        EXPLODE(
            FROM_JSON(
                details:flow_progress:data_quality:expectations,
                'ARRAY<STRUCT<name: STRING, passed_records: BIGINT, failed_records: BIGINT>>'
            )
        ) AS exp
    FROM event_log("pipeline-id-here")
    WHERE event_type = 'flow_progress'
        AND details:flow_progress:data_quality IS NOT NULL
)
SELECT
    week_start,
    flow_name,
    exp.name AS rule_name,
    SUM(exp.passed_records) AS total_passed,
    SUM(exp.failed_records) AS total_failed,
    ROUND(
        SUM(exp.passed_records) * 100.0 /
        NULLIF(SUM(exp.passed_records) + SUM(exp.failed_records), 0), 2
    ) AS pass_rate_pct,
    -- 전주 대비 변화
    LAG(
        ROUND(SUM(exp.passed_records) * 100.0 /
        NULLIF(SUM(exp.passed_records) + SUM(exp.failed_records), 0), 2)
    ) OVER (PARTITION BY flow_name, exp.name ORDER BY week_start) AS prev_week_rate
FROM weekly_quality
GROUP BY week_start, flow_name, exp.name
ORDER BY week_start DESC, flow_name;

-- 각 flow별 평균 처리 시간과 처리량 (병목 구간 식별)
SELECT
    details:flow_progress:flow_name AS flow_name,
    COUNT(*) AS total_runs,
    ROUND(AVG(CAST(details:flow_progress:metrics:num_output_rows AS BIGINT)), 0) AS avg_rows,
    ROUND(AVG(
        TIMESTAMPDIFF(SECOND,
            LAG(timestamp) OVER (PARTITION BY details:flow_progress:flow_name ORDER BY timestamp),
            timestamp
        )
    ), 1) AS avg_interval_sec
FROM event_log("pipeline-id-here")
WHERE event_type = 'flow_progress'
    AND details:flow_progress:status = 'COMPLETED'
    AND timestamp > CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY details:flow_progress:flow_name
ORDER BY avg_rows DESC;

-- 파이프라인 건강 상태 종합 뷰
CREATE OR REPLACE VIEW catalog.schema.v_pipeline_health AS
SELECT
    'error_count' AS metric_name,
    COUNT(*) AS metric_value,
    CURRENT_TIMESTAMP() AS checked_at
FROM event_log("pipeline-id-here")
WHERE level = 'ERROR'
    AND DATE(timestamp) = CURRENT_DATE()

UNION ALL

SELECT
    'avg_duration_min',
    ROUND(AVG(TIMESTAMPDIFF(MINUTE, start_time, end_time)), 1),
    CURRENT_TIMESTAMP()
FROM (
    SELECT
        origin.update_id,
        MIN(timestamp) AS start_time,
        MAX(timestamp) AS end_time
    FROM event_log("pipeline-id-here")
    WHERE DATE(timestamp) = CURRENT_DATE()
    GROUP BY origin.update_id
);

# Lakeflow Jobs로 매일 이벤트 로그를 장기 보관 테이블에 적재
# pipeline_event_archive 테이블은 별도로 생성해 두어야 합니다

spark.sql("""
    INSERT INTO catalog.schema.pipeline_event_archive
    SELECT
        *,
        CURRENT_TIMESTAMP() AS archived_at
    FROM event_log("pipeline-id-here")
    WHERE DATE(timestamp) = CURRENT_DATE() - INTERVAL 1 DAY
""")

# 장기 보관 테이블에서 월간 리포트 생성 가능
# 30일 기본 보존 기간 이후에도 이력을 추적할 수 있습니다

-- 이벤트 로그를 장기 보관 테이블로 복사
INSERT INTO catalog.schema.pipeline_event_archive
SELECT * FROM event_log("pipeline-id-here")
WHERE DATE(timestamp) = CURRENT_DATE() - INTERVAL 1 DAY;