SDP 이벤트 로그 기초

-- 파이프라인 ID로 이벤트 로그 조회
SELECT * FROM event_log("pipeline-id-here")
ORDER BY timestamp DESC
LIMIT 100;

-- 파이프라인 이름으로 조회
SELECT * FROM event_log(TABLE(catalog.schema.my_table))
ORDER BY timestamp DESC;

-- 각 테이블별 처리 행 수와 소요 시간 조회
SELECT
    timestamp,
    details:flow_progress:data_quality:expectations AS expectations,
    details:flow_progress:metrics:num_output_rows AS output_rows,
    details:flow_progress:status AS status,
    details:flow_progress:flow_name AS flow_name
FROM event_log("pipeline-id-here")
WHERE event_type = 'flow_progress'
    AND details:flow_progress:status = 'COMPLETED'
ORDER BY timestamp DESC;

-- Expectations 결과만 추출
WITH expectations_data AS (
    SELECT
        timestamp,
        details:flow_progress:flow_name AS flow_name,
        EXPLODE(
            FROM_JSON(
                details:flow_progress:data_quality:expectations,
                'ARRAY<STRUCT<name: STRING, dataset: STRING, passed_records: BIGINT, failed_records: BIGINT>>'
            )
        ) AS expectation
    FROM event_log("pipeline-id-here")
    WHERE event_type = 'flow_progress'
        AND details:flow_progress:data_quality IS NOT NULL
)
SELECT
    DATE(timestamp) AS date,
    flow_name,
    expectation.name AS rule_name,
    expectation.passed_records,
    expectation.failed_records,
    ROUND(
        expectation.passed_records * 100.0 /
        (expectation.passed_records + expectation.failed_records), 2
    ) AS pass_rate_pct
FROM expectations_data
ORDER BY timestamp DESC;

-- 오류 이벤트만 필터링
SELECT
    timestamp,
    level,
    message,
    error.exceptions AS error_details
FROM event_log("pipeline-id-here")
WHERE level = 'ERROR'
ORDER BY timestamp DESC
LIMIT 20;

-- 파이프라인 실행별 소요 시간 분석
WITH update_events AS (
    SELECT
        origin.update_id,
        MIN(CASE WHEN event_type = 'user_action' AND details:user_action:action = 'START' THEN timestamp END) AS start_time,
        MAX(CASE WHEN event_type = 'user_action' AND details:user_action:action IN ('STOP', 'COMPLETE') THEN timestamp END) AS end_time
    FROM event_log("pipeline-id-here")
    GROUP BY origin.update_id
)
SELECT
    update_id,
    start_time,
    end_time,
    TIMESTAMPDIFF(MINUTE, start_time, end_time) AS duration_minutes
FROM update_events
WHERE start_time IS NOT NULL AND end_time IS NOT NULL
ORDER BY start_time DESC
LIMIT 10;

-- 일별 테이블별 처리 행 수 추이
SELECT
    DATE(timestamp) AS date,
    details:flow_progress:flow_name AS table_name,
    SUM(CAST(details:flow_progress:metrics:num_output_rows AS BIGINT)) AS total_rows
FROM event_log("pipeline-id-here")
WHERE event_type = 'flow_progress'
    AND details:flow_progress:status = 'COMPLETED'
GROUP BY 1, 2
ORDER BY date DESC, table_name;

-- 데이터 품질 모니터링 뷰
CREATE OR REPLACE VIEW catalog.schema.v_pipeline_quality AS
WITH latest_expectations AS (
    SELECT
        timestamp,
        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
        AND DATE(timestamp) = CURRENT_DATE()
)
SELECT
    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
FROM latest_expectations
GROUP BY flow_name, exp.name;