수집 파이프라인 구성

CREATE CONNECTION mysql_prod
TYPE MYSQL
OPTIONS (
    host = 'mysql-prod.example.com',    -- MySQL 서버 주소
    port = '3306',                       -- 포트 (기본값: 3306)
    user = 'databricks_reader',          -- 접속 계정
    password = secret('scope', 'mysql-password'),  -- Secret 참조
    useSSL = 'true',                     -- SSL 암호화 사용
    trustServerCertificate = 'false'     -- 서버 인증서 검증
);

CREATE CONNECTION postgres_analytics
TYPE POSTGRESQL
OPTIONS (
    host = 'postgres.example.com',
    port = '5432',
    user = 'databricks_cdc',
    password = secret('scope', 'pg-password'),
    database = 'analytics',              -- 대상 데이터베이스 지정
    sslmode = 'require'                  -- SSL 모드 (require/verify-full)
);

CREATE CONNECTION sqlserver_erp
TYPE SQLSERVER
OPTIONS (
    host = 'sqlserver.example.com',
    port = '1433',
    user = 'databricks_reader',
    password = secret('scope', 'mssql-password'),
    database = 'ERP',
    encrypt = 'true',
    trustServerCertificate = 'false'
);

CREATE CONNECTION salesforce_crm
TYPE SALESFORCE
OPTIONS (
    sfUrl = 'https://mycompany.my.salesforce.com',
    oauthClientId = secret('scope', 'sf-client-id'),
    oauthClientSecret = secret('scope', 'sf-client-secret'),
    oauthRefreshToken = secret('scope', 'sf-refresh-token')
);

# 방법 1: 전체 스키마 수집
@dlt.table
def ingest_all():
    return mysql.read_changefeed(
        connection="mysql_prod",
        source_schema="ecommerce",
        # source_table 생략 시 전체 스키마 수집
    )

# 방법 2: 특정 테이블만 선택
@dlt.table
def ingest_customers():
    return mysql.read_changefeed(
        connection="mysql_prod",
        source_schema="ecommerce",
        source_table="customers"
    )

소스: <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;