SDP란? — 선언적 파이프라인

# ❌ 명령형: 모든 것을 직접 관리해야 합니다
df = spark.readStream.format("delta").load("/bronze/orders")
cleaned = df.filter(col("order_id").isNotNull()).dropDuplicates(["order_id"])
query = (cleaned.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/silver_orders")
    .option("mergeSchema", "true")
    .outputMode("append")
    .trigger(availableNow=True)
    .toTable("silver_orders"))
query.awaitTermination()

-- ✅ 선언형 (SDP): 결과만 정의합니다. 나머지는 시스템이 알아서 합니다.
CREATE OR REFRESH STREAMING TABLE silver_orders (
    CONSTRAINT valid_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW
)
AS SELECT * FROM STREAM(bronze_orders);

-- Streaming Table
CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files('s3://bucket/orders/', format => 'json');

-- Materialized View
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_revenue
AS SELECT DATE(order_date) AS day, SUM(amount) AS revenue
FROM silver_orders GROUP BY 1;

from pyspark import pipelines as dp

@dp.table(comment="Bronze 주문 데이터")
def bronze_orders():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("s3://bucket/orders/"))

@dp.materialized_view(comment="일별 매출 집계")
def gold_daily_revenue():
    return spark.sql("""
        SELECT DATE(order_date) AS day, SUM(amount) AS revenue
        FROM silver_orders GROUP BY 1
    """)

resources:
  pipelines:
    sales_pipeline:
      name: "sales-medallion-pipeline"
      catalog: "production"
      schema: "ecommerce"
      serverless: true
      continuous: false
      libraries:
        - notebook:
            path: "/Workspace/pipelines/bronze.sql"
        - notebook:
            path: "/Workspace/pipelines/silver.sql"
        - notebook:
            path: "/Workspace/pipelines/gold.sql"
      configuration:
        "pipelines.trigger.interval": "1 hour"

-- 파이프라인 이벤트 로그 조회
SELECT
    timestamp,
    event_type,
    message,
    details
FROM event_log(TABLE(catalog.schema.my_streaming_table))
WHERE event_type IN ('flow_progress', 'dataset_quality')
ORDER BY timestamp DESC;

SDP (변환) + Lakeflow Jobs (오케스트레이션) = Databricks 네이티브 스택
SDP (변환) + Airflow (오케스트레이션) = 하이브리드 스택 (기존 Airflow 투자 보존)

시나리오: 일 4회 실행, 회당 15분 소요, 10 DBU/시간

Triggered 모드:
  4회 × 15분 × 10 DBU/시간 = 10 DBU/일 = 300 DBU/월

Continuous 모드:
  24시간 × 10 DBU/시간 = 240 DBU/일 = 7,200 DBU/월
  (→ Triggered 대비 24배 비용!)

손익 분기점: 하루 약 16회 이상 실행 시 Continuous가 유리
  (단, 실시간 처리의 비즈니스 가치를 별도 평가해야 합니다)

configuration:
  "pipelines.trigger.interval": "5 minutes"   # 5분마다 새 데이터 확인

서버리스 SDP를 선택하세요:
  ✅ 파이프라인 실행 시간이 짧은 경우 (< 2시간)
  ✅ 간헐적으로 실행되는 배치 ETL
  ✅ 특별한 라이브러리/네트워크 요구사항이 없는 경우
  ✅ 빠른 개발 반복이 필요한 경우

Classic SDP를 선택하세요:
  ✅ 외부 네트워크 접근이 필요한 경우 (DB 직접 연결 등)
  ✅ GPU가 필요한 ML 파이프라인
  ✅ 커스텀 C 라이브러리가 필요한 경우
  ✅ 매우 긴 실행 시간 (> 8시간, Spot 인스턴스 활용 시 비용 유리)

-- 단일 규칙: NULL 체크
CONSTRAINT valid_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW

-- 비율 기반 규칙: 95% 이상 통과해야 함
CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW

-- 치명적 규칙: 위반 시 파이프라인 실패
CONSTRAINT critical_date EXPECT (order_date IS NOT NULL) ON VIOLATION FAIL UPDATE

-- 복합 규칙
CONSTRAINT valid_status EXPECT (status IN ('pending', 'completed', 'cancelled'))
    ON VIOLATION DROP ROW