Streaming Table 증분 처리

# 전체 테이블 Full Refresh
curl -X POST "https://<workspace-url>/api/2.0/pipelines/<pipeline-id>/updates" \
  -H "Authorization: Bearer <token>" \
  -d '{"full_refresh": true}'

# 특정 테이블만 Full Refresh
curl -X POST "https://<workspace-url>/api/2.0/pipelines/<pipeline-id>/updates" \
  -H "Authorization: Bearer <token>" \
  -d '{
    "full_refresh_selection": ["catalog.schema.target_table"]
  }'

# 전체 Full Refresh
databricks pipelines start-update <pipeline-id> --full-refresh

# 특정 테이블만 Full Refresh
databricks pipelines start-update <pipeline-id> \
  --full-refresh-selection "catalog.schema.target_table"

CREATE OR REFRESH STREAMING TABLE orders
TBLPROPERTIES ('pipelines.reset.allowed' = 'false')
AS SELECT * FROM STREAM(raw_orders);

@dlt.table(
    table_properties={"pipelines.reset.allowed": "false"}
)
def orders():
    return spark.readStream.table("raw_orders")

-- 대상 테이블 정의 (스키마만, 소스 없음)
CREATE OR REFRESH STREAMING TABLE all_orders (
  order_id BIGINT,
  product STRING,
  amount DECIMAL(10,2),
  region STRING,
  order_date DATE
);

-- Flow 1: 한국 주문
CREATE FLOW korea_orders
AS INSERT INTO all_orders BY NAME
SELECT * FROM STREAM(raw_korea_orders);

-- Flow 2: 미국 주문 (나중에 추가 — 기존 Flow에 영향 없음)
CREATE FLOW us_orders
AS INSERT INTO all_orders BY NAME
SELECT * FROM STREAM(raw_us_orders);

-- 과거 데이터를 한 번만 적재 (배치 소스 사용 가능)
CREATE FLOW backfill_legacy_orders
  ONCE
AS INSERT INTO all_orders BY NAME
SELECT
  order_id,
  product,
  amount,
  region,
  order_date
FROM legacy_orders_archive  -- STREAM 없이 배치로 읽기
WHERE order_date < '2025-01-01';

import dlt
from pyspark.sql.functions import *

# 대상 테이블 정의
dlt.create_streaming_table("all_orders")

# Flow 1: 한국 주문
@dlt.append_flow(target="all_orders")
def korea_orders():
    return spark.readStream.table("raw_korea_orders")

# Flow 2: 미국 주문
@dlt.append_flow(target="all_orders")
def us_orders():
    return spark.readStream.table("raw_us_orders")

# 일회성 백필
@dlt.append_flow(target="all_orders", once=True)
def backfill_legacy():
    return spark.read.table("legacy_orders_archive") \
        .filter("order_date < '2025-01-01'")

CREATE OR REFRESH STREAMING TABLE raw_events
AS SELECT *
FROM cloud_files(
  's3://bucket/events/',
  'json',
  map(
    'cloudFiles.modifiedAfter', '2026-01-01T00:00:00.000Z',
    'cloudFiles.inferColumnTypes', 'true'
  )
);

@dlt.table
def processed_events():
    return spark.readStream \
        .option("startingVersion", 42) \
        .table("catalog.schema.raw_events")