Delta Live Tables: 선언적 ETL 파이프라인의 새로운 기준

CREATE STREAMING TABLE orders_silver_cleaned
AS
SELECT *
FROM STREAM(LIVE.bronze_orders)
WHERE order_status IS NOT NULL;

CREATE STREAMING TABLE orders_silver_cleaned (
  CONSTRAINT valid_order_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valid_amount   EXPECT (amount > 0)           ON VIOLATION DROP ROW,
  CONSTRAINT valid_customer EXPECT (customer_id IS NOT NULL) ON VIOLATION WARN
)
AS
SELECT
  orderid     AS order_id,
  orderdate   AS order_date,
  customerid  AS customer_id,
  totalamount AS amount,
  status
FROM STREAM(LIVE.orders_bronze);

-- Auto Loader를 사용한 클라우드 스토리지 수집
CREATE STREAMING TABLE orders_bronze
AS
SELECT
  *,
  _metadata.file_name AS filename,
  current_timestamp() AS load_time
FROM cloud_files(
  '/Volumes/catalog/schema/landing_volume/orders',
  'csv',
  map('cloudFiles.inferColumnTypes', 'true')
);

-- 데이터 품질 기대값이 포함된 Silver 테이블
CREATE STREAMING TABLE orders_silver_cleaned (
  CONSTRAINT valid_order   EXPECT (order_id IS NOT NULL)   ON VIOLATION DROP ROW,
  CONSTRAINT valid_customer EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
)
AS
SELECT
  orderid     AS order_id,
  orderdate   AS order_date,
  customerid  AS customer_id,
  totalamount AS total_amount,
  status,
  filename    AS file_name,
  load_time
FROM STREAM(LIVE.orders_bronze);

-- SCD Type 2: 고객 이력 추적
CREATE STREAMING TABLE customers_silver;

APPLY CHANGES INTO LIVE.customers_silver
FROM STREAM(LIVE.customers_silver_cleaned)
KEYS (customer_id)
SEQUENCE BY load_time
STORED AS SCD TYPE 2;

-- 도시별 매출 집계 Materialized View
CREATE MATERIALIZED VIEW city_wise_sales_gold
AS
SELECT
  c.city,
  SUM(o.total_amount) AS total_sales,
  COUNT(DISTINCT o.order_id) AS order_count
FROM LIVE.orders_silver  AS o
JOIN LIVE.customers_silver AS c ON o.customer_id = c.customer_id
GROUP BY c.city;

from pyspark import pipelines as dp
from pyspark.sql.functions import current_timestamp, col, sum as _sum

# 소스 경로 설정 (파이프라인 파라미터로 관리)
source_path = spark.conf.get("source_path")

# Bronze: 주문 데이터 수집
@dp.table(name="orders_bronze")
def orders_bronze():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "csv")
            .option("cloudFiles.inferColumnTypes", "true")
            .load(f"{source_path}/orders")
            .withColumn("filename", col("_metadata.file_path"))
            .withColumn("load_time", current_timestamp())
    )

# Silver: 데이터 품질 기대값 적용
@dp.table()
@dp.expect_or_drop("valid_order_id", "order_id is not null")
@dp.expect_or_drop("valid_customer_id", "customer_id is not null")
def orders_silver_cleaned():
    return (
        spark.readStream.table("orders_bronze")
            .selectExpr(
                "orderid     AS order_id",
                "orderdate   AS order_date",
                "customerid  AS customer_id",
                "totalamount AS total_amount",
                "status",
                "filename AS file_name",
                "load_time"
            )
    )

# Gold: 도시별 매출 Materialized View
@dp.materialized_view()
def city_wise_sales_gold():
    orders    = spark.read.table("orders_silver")
    customers = spark.read.table("customers_silver")
    return (
        orders.join(customers, "customer_id")
              .groupBy("city")
              .agg(_sum("total_amount").alias("total_sales"))
    )

-- AUTO CDC를 사용한 SCD Type 2
CREATE STREAMING TABLE customers_silver;

CREATE FLOW customer_silver_flow AS AUTO CDC
INTO customers_silver
FROM STREAM(customers_silver_cleaned)
KEYS (customer_id)
SEQUENCE BY load_time
STORED AS SCD TYPE 2;