왜 데이터 품질 관리가 필요한가?과 Expectations란?

CREATE OR REFRESH STREAMING TABLE silver_orders (
    -- 경고만 (모니터링용)
    CONSTRAINT valid_email
        EXPECT (email RLIKE '^[^@]+@[^@]+\\.[^@]+$')
        ON VIOLATION WARN,

    -- 위반 행 삭제 (품질 보장)
    CONSTRAINT valid_order_id
        EXPECT (order_id IS NOT NULL)
        ON VIOLATION DROP ROW,

    CONSTRAINT positive_amount
        EXPECT (amount > 0)
        ON VIOLATION DROP ROW,

    -- 파이프라인 중지 (치명적 오류)
    CONSTRAINT valid_date
        EXPECT (order_date >= '2020-01-01')
        ON VIOLATION FAIL UPDATE
)
AS SELECT * FROM STREAM(bronze_orders);

import dlt
from pyspark.sql.functions import col, expr

# 경고만 (WARN)
@dlt.table(comment="정제된 주문 데이터")
@dlt.expect("valid_order_id", "order_id IS NOT NULL")
@dlt.expect("valid_email", "email RLIKE '^[^@]+@[^@]+\\\\.[^@]+$'")
def silver_orders():
    return (
        dlt.read_stream("bronze_orders")
           .select(
               col("order_id").cast("bigint"),
               col("amount").cast("decimal(12,2)"),
               col("order_date").cast("timestamp"),
               col("email")
           )
    )

# 행 삭제 (DROP)
@dlt.table()
@dlt.expect_or_drop("positive_amount", "amount > 0")
@dlt.expect_or_drop("valid_status", "status IN ('pending', 'completed', 'cancelled')")
def silver_orders_clean():
    return dlt.read_stream("silver_orders")

# 파이프라인 중지 (FAIL)
@dlt.table()
@dlt.expect_or_fail("valid_date", "order_date >= '2020-01-01'")
def silver_orders_strict():
    return dlt.read_stream("silver_orders")