파이프라인 오케스트레이션 패턴

# Lakeflow Jobs SDK로 위 파이프라인 생성
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import *

w = WorkspaceClient()

w.jobs.create(
    name="daily-data-pipeline",
    tasks=[
        Task(
            task_key="ingest",
            pipeline_task=PipelineTask(pipeline_id="<ingest-pipeline-id>"),
        ),
        Task(
            task_key="transform",
            depends_on=[TaskDependency(task_key="ingest")],
            pipeline_task=PipelineTask(pipeline_id="<sdp-pipeline-id>"),
        ),
        Task(
            task_key="quality_check",
            depends_on=[TaskDependency(task_key="transform")],
            sql_task=SqlTask(
                query=SqlTaskQuery(query_id="<quality-check-query-id>"),
                warehouse_id="<warehouse-id>"
            ),
        ),
        Task(
            task_key="update_features",
            depends_on=[TaskDependency(task_key="quality_check")],
            notebook_task=NotebookTask(
                notebook_path="/Workspace/pipelines/update_features"
            ),
        ),
        Task(
            task_key="refresh_dashboard",
            depends_on=[TaskDependency(task_key="quality_check")],
            sql_task=SqlTask(
                query=SqlTaskQuery(query_id="<refresh-mv-query-id>"),
                warehouse_id="<warehouse-id>"
            ),
        ),
    ],
    schedule=CronSchedule(
        quartz_cron_expression="0 0 6 * * ?",  # 매일 06:00
        timezone_id="Asia/Seoul"
    ),
    email_notifications=JobEmailNotifications(
        on_failure=["data-team@company.com"]
    )
)

-- SDP에서 SCD Type 1 구현
CREATE OR REFRESH STREAMING TABLE silver_customers;

APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_customers)
KEYS (customer_id)
SEQUENCE BY _commit_timestamp
STORED AS SCD TYPE 1;

-- SDP에서 SCD Type 2 구현
CREATE OR REFRESH STREAMING TABLE silver_customers_history;

APPLY CHANGES INTO silver_customers_history
FROM STREAM(bronze_customers)
KEYS (customer_id)
SEQUENCE BY _commit_timestamp
STORED AS SCD TYPE 2;

-- 결과 테이블 구조:
-- | customer_id | name   | city  | __START_AT     | __END_AT       |
-- |-------------|--------|-------|----------------|----------------|
-- | 1001        | 김철수 | 서울  | 2025-01-01     | 2025-03-15     |
-- | 1001        | 김철수 | 부산  | 2025-03-15     | NULL           | ← 현재 값

-- 같은 Bronze 소스에서 두 가지 Silver 테이블 생성

-- Silver 1: 현재 상태 (조인용, 대시보드용) — SCD Type 1
CREATE OR REFRESH STREAMING TABLE silver_customers_current;
APPLY CHANGES INTO silver_customers_current
FROM STREAM(bronze_customers)
KEYS (customer_id) SEQUENCE BY _commit_timestamp
STORED AS SCD TYPE 1;

-- Silver 2: 전체 이력 (감사, 시계열 분석용) — SCD Type 2
CREATE OR REFRESH STREAMING TABLE silver_customers_history;
APPLY CHANGES INTO silver_customers_history
FROM STREAM(bronze_customers)
KEYS (customer_id) SEQUENCE BY _commit_timestamp
STORED AS SCD TYPE 2;

-- Gold: 현재 상태 기반 집계 (Type 1에서 읽음 → 빠른 조인)
CREATE OR REFRESH MATERIALIZED VIEW gold_customer_revenue AS
SELECT c.customer_id, c.name, c.city,
       COUNT(o.order_id) AS total_orders,
       SUM(o.amount) AS total_revenue
FROM silver_customers_current c
JOIN silver_orders o ON c.customer_id = o.customer_id
GROUP BY c.customer_id, c.name, c.city;

-- 감사 리포트: 이력 기반 조회 (Type 2에서 읽음)
-- "2025년 1분기에 서울에 거주했던 고객 목록"
SELECT * FROM silver_customers_history
WHERE city = '서울'
  AND __START_AT <= '2025-03-31'
  AND (__END_AT > '2025-01-01' OR __END_AT IS NULL);

-- Silver 테이블에 CDF 활성화 (다운스트림 전파용)
ALTER TABLE silver_customers
SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

# Silver의 변경사항을 외부 시스템에 Reverse ETL
def push_to_crm(batch_df, batch_id):
    updates = batch_df.filter("_change_type IN ('insert', 'update_postimage')")
    # CRM API 호출로 고객 정보 동기화
    for row in updates.collect():
        crm_api.update_customer(row.customer_id, row.name, row.email)

(spark.readStream
    .format("delta")
    .option("readChangeFeed", "true")
    .table("silver_customers")
    .writeStream
    .foreachBatch(push_to_crm)
    .option("checkpointLocation", "/checkpoints/reverse-etl")
    .trigger(availableNow=True)
    .start()
)