수집 파이프라인 구성 — 심화

소스: <connection>.<source_schema>.<source_table>
타겟: <target_catalog>.<target_schema>.<source_table>

예시:
소스: mysql_prod.ecommerce.customers
타겟: analytics.bronze.customers

-- 메타데이터 컬럼 활용 예시
SELECT
    _change_type,
    _commit_timestamp,
    customer_id,
    name
FROM analytics.bronze.customers
WHERE _change_type = 'UPDATE'
  AND _commit_timestamp >= CURRENT_DATE - INTERVAL 1 DAY
ORDER BY _commit_timestamp DESC;

-- Silver 테이블에서 Liquid Clustering 적용 (권장)
CREATE OR REFRESH STREAMING TABLE analytics.silver.orders
CLUSTER BY (order_date, customer_id)
AS SELECT * FROM STREAM(analytics.bronze.orders);

-- 기존 테이블에 Liquid Clustering 추가
ALTER TABLE analytics.silver.orders
CLUSTER BY (order_date, customer_id);

-- 파이프라인 실행 이력 조회 (시스템 테이블 활용)
SELECT
    pipeline_id,
    update_id,
    state,
    creation_time,
    TIMESTAMPDIFF(MINUTE, creation_time, end_time) AS duration_minutes
FROM system.lakeflow.pipeline_event_log
WHERE pipeline_id = '<pipeline-id>'
ORDER BY creation_time DESC
LIMIT 20;

-- 일별 수집 건수 추이
SELECT
    DATE(_ingested_at) AS ingest_date,
    COUNT(*) AS records_ingested,
    COUNT(DISTINCT _source_table) AS tables_updated
FROM analytics.bronze.orders
GROUP BY DATE(_ingested_at)
ORDER BY ingest_date DESC
LIMIT 30;

-- SQL Alert 예시: 수집 지연이 30분 이상일 때 알림
-- (Databricks SQL Alert로 등록)
SELECT
    MAX(TIMESTAMPDIFF(MINUTE, _commit_timestamp, _ingested_at)) AS max_latency_min
FROM analytics.bronze.orders
WHERE _ingested_at >= CURRENT_TIMESTAMP - INTERVAL 1 HOUR
HAVING max_latency_min > 30;

# Lakeflow Jobs로 파이프라인 스케줄링 (API 예시)
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# 매 2시간마다 파이프라인 트리거
w.jobs.create(
    name="ecommerce_ingestion_schedule",
    tasks=[{
        "task_key": "trigger_ingestion",
        "pipeline_task": {
            "pipeline_id": "<pipeline-id>",
            "full_refresh": False
        }
    }],
    schedule={
        "quartz_cron_expression": "0 0 */2 * * ?",
        "timezone_id": "Asia/Seoul"
    }
)

-- Bronze 테이블에 보관 기간 설정 (예: 90일)
ALTER TABLE analytics.bronze.events
SET TBLPROPERTIES (
    'delta.deletedFileRetentionDuration' = 'interval 90 days',
    'delta.logRetentionDuration' = 'interval 90 days'
);

-- 오래된 데이터 정리 (VACUUM)
VACUUM analytics.bronze.events RETAIN 90 HOURS;