Data Sync

-- Lakebase 데이터베이스 생성 (이미 생성되어 있다면 건너뜁니다)
CREATE DATABASE my_catalog.my_schema.ecommerce_db
TYPE LAKEBASE;

-- Lakebase에 테이블 생성
-- psql 또는 Python 클라이언트로 접속하여 실행합니다
CREATE TABLE orders (
    order_id    SERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    product     VARCHAR(200) NOT NULL,
    amount      DECIMAL(10,2) NOT NULL,
    status      VARCHAR(20) DEFAULT 'pending',
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,
    name        VARCHAR(100) NOT NULL,
    email       VARCHAR(200) UNIQUE NOT NULL,
    tier        VARCHAR(20) DEFAULT 'standard',
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- orders 테이블의 CONTINUOUS 동기화 설정
CREATE SYNC my_catalog.my_schema.orders_sync
SOURCE my_catalog.my_schema.ecommerce_db.orders
TARGET my_catalog.my_schema.orders_delta
MODE CONTINUOUS;

-- customers 테이블의 TRIGGERED 동기화 설정 (비용 절감)
CREATE SYNC my_catalog.my_schema.customers_sync
SOURCE my_catalog.my_schema.ecommerce_db.customers
TARGET my_catalog.my_schema.customers_delta
MODE TRIGGERED;

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# CONTINUOUS 동기화 생성
sync = w.lakebase.create_sync(
    name="orders_sync",
    source_database="my_catalog.my_schema.ecommerce_db",
    source_table="orders",
    target_catalog="my_catalog",
    target_schema="my_schema",
    target_table="orders_delta",
    mode="CONTINUOUS"
)

print(f"동기화 생성 완료: {sync.name}")
print(f"상태: {sync.state}")

-- TRIGGERED 모드의 동기화를 수동으로 실행합니다
ALTER SYNC my_catalog.my_schema.customers_sync TRIGGER;

# Python SDK로 트리거 실행
w.lakebase.trigger_sync(name="customers_sync")

-- 스키마 변경 전: 동기화 일시 중지
ALTER SYNC my_catalog.my_schema.orders_sync PAUSE;

-- Lakebase에서 스키마 변경 수행
-- (psql 등으로 접속하여 ALTER TABLE 실행)

-- 스키마 변경 후: 동기화 재개
ALTER SYNC my_catalog.my_schema.orders_sync RESUME;

-- 동기화 상태 조회
DESCRIBE SYNC my_catalog.my_schema.orders_sync;

-- 결과 예시:
-- name: orders_sync
-- state: RUNNING
-- mode: CONTINUOUS
-- source: my_catalog.my_schema.ecommerce_db.orders
-- target: my_catalog.my_schema.orders_delta
-- last_sync_time: 2025-06-15T10:30:45Z
-- rows_synced: 1,234,567
-- lag_seconds: 3

# Python SDK로 상태 확인
sync_status = w.lakebase.get_sync(name="orders_sync")

print(f"상태: {sync_status.state}")
print(f"마지막 동기화: {sync_status.last_sync_time}")
print(f"동기화된 행 수: {sync_status.rows_synced}")
print(f"지연시간(초): {sync_status.lag_seconds}")

-- 기존 동기화 삭제
DROP SYNC my_catalog.my_schema.orders_sync;

-- SNAPSHOT 모드로 전체 재동기화 (초기 로드)
CREATE SYNC my_catalog.my_schema.orders_sync
SOURCE my_catalog.my_schema.ecommerce_db.orders
TARGET my_catalog.my_schema.orders_delta
MODE SNAPSHOT;

-- 초기 로드 완료 후 CONTINUOUS로 변경
ALTER SYNC my_catalog.my_schema.orders_sync
SET MODE CONTINUOUS;

-- Delta Lake에서 실시간 매출 분석 쿼리 (DBSQL에서 실행)
SELECT
    DATE_TRUNC('hour', created_at) AS order_hour,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value
FROM my_catalog.my_schema.orders_delta
WHERE created_at >= CURRENT_DATE
GROUP BY 1
ORDER BY 1 DESC;