데이터 품질 관리 — 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")

CREATE OR REFRESH STREAMING TABLE silver_transactions (
    -- 여러 조건을 AND로 결합
    CONSTRAINT valid_transaction
        EXPECT (
            transaction_id IS NOT NULL
            AND amount > 0
            AND amount < 1000000
            AND currency IN ('USD', 'EUR', 'KRW', 'JPY')
        )
        ON VIOLATION DROP ROW,

    -- OR 조건으로 유연한 검증
    CONSTRAINT valid_contact
        EXPECT (
            email IS NOT NULL
            OR phone IS NOT NULL
        )
        ON VIOLATION WARN,

    -- CASE 표현식을 활용한 조건부 검증
    CONSTRAINT valid_discount
        EXPECT (
            CASE
                WHEN customer_type = 'premium' THEN discount <= 0.30
                WHEN customer_type = 'regular' THEN discount <= 0.15
                ELSE discount <= 0.05
            END
        )
        ON VIOLATION DROP ROW
)
AS SELECT * FROM STREAM(bronze_transactions);

import dlt

# 여러 Expectation을 딕셔너리로 정의
order_expectations = {
    "valid_order_id": "order_id IS NOT NULL",
    "positive_amount": "amount > 0 AND amount < 1000000",
    "valid_status": "status IN ('pending', 'completed', 'cancelled')",
    "valid_date": "order_date >= '2020-01-01'"
}

# expect_all: 모든 규칙에 WARN 적용
@dlt.table()
@dlt.expect_all(order_expectations)
def silver_orders_monitored():
    return dlt.read_stream("bronze_orders")

# expect_all_or_drop: 모든 규칙에 DROP 적용
@dlt.table()
@dlt.expect_all_or_drop(order_expectations)
def silver_orders_clean():
    return dlt.read_stream("bronze_orders")

# expect_all_or_fail: 모든 규칙에 FAIL 적용
@dlt.table()
@dlt.expect_all_or_fail(order_expectations)
def silver_orders_strict():
    return dlt.read_stream("bronze_orders")

-- 1단계: 모든 데이터를 일단 받되, 품질 규칙은 WARN으로 설정
CREATE OR REFRESH STREAMING TABLE bronze_orders_validated (
    CONSTRAINT valid_order_id
        EXPECT (order_id IS NOT NULL)
        ON VIOLATION WARN,
    CONSTRAINT positive_amount
        EXPECT (amount > 0)
        ON VIOLATION WARN,
    CONSTRAINT valid_email
        EXPECT (email RLIKE '^[^@]+@[^@]+\\.[^@]+$')
        ON VIOLATION WARN
)
AS SELECT * FROM STREAM(raw_orders);

-- 2단계: 정상 데이터만 Silver로 이동
CREATE OR REFRESH STREAMING TABLE silver_orders_clean
AS SELECT *
FROM STREAM(bronze_orders_validated)
WHERE order_id IS NOT NULL
  AND amount > 0
  AND email RLIKE '^[^@]+@[^@]+\\.[^@]+$';

-- 3단계: 위반 데이터를 격리 테이블로 이동
CREATE OR REFRESH STREAMING TABLE quarantine_orders
AS SELECT
    *,
    CASE
        WHEN order_id IS NULL THEN 'missing_order_id'
        WHEN amount <= 0 THEN 'invalid_amount'
        WHEN NOT email RLIKE '^[^@]+@[^@]+\\.[^@]+$' THEN 'invalid_email'
    END AS violation_reason,
    current_timestamp() AS quarantined_at
FROM STREAM(bronze_orders_validated)
WHERE order_id IS NULL
   OR amount <= 0
   OR NOT email RLIKE '^[^@]+@[^@]+\\.[^@]+$';

import dlt
from pyspark.sql import SparkSession

def get_rules_for_table(table_name):
    """외부 규칙 테이블에서 해당 테이블의 품질 규칙을 조회합니다."""
    spark = SparkSession.builder.getOrCreate()
    rules = spark.sql(f"""
        SELECT rule_name, rule_expression
        FROM catalog.schema.data_quality_rules
        WHERE target_table = '{table_name}'
          AND is_active = true
    """).collect()
    return {row.rule_name: row.rule_expression for row in rules}

@dlt.table()
@dlt.expect_all_or_drop(get_rules_for_table("silver_orders"))
def silver_orders():
    return dlt.read_stream("bronze_orders")

-- 파이프라인 이벤트 로그에서 Expectation 결과 조회
SELECT
    timestamp,
    details:flow_definition.output_dataset AS dataset_name,
    details:flow_progress.data_quality.expectations AS expectations
FROM event_log(TABLE(my_catalog.my_schema.my_pipeline))
WHERE event_type = 'flow_progress'
  AND details:flow_progress.data_quality IS NOT NULL
ORDER BY timestamp DESC;

{
  "expectations": [
    {
      "name": "valid_order_id",
      "dataset": "silver_orders",
      "passed_records": 9850,
      "failed_records": 150
    },
    {
      "name": "positive_amount",
      "dataset": "silver_orders",
      "passed_records": 9990,
      "failed_records": 10
    }
  ]
}

-- 일별 데이터 품질 추이 집계
CREATE OR REFRESH MATERIALIZED VIEW data_quality_daily_summary
AS
SELECT
    DATE(timestamp) AS check_date,
    details:flow_definition.output_dataset AS dataset_name,
    exp.name AS expectation_name,
    SUM(exp.passed_records) AS total_passed,
    SUM(exp.failed_records) AS total_failed,
    ROUND(
        SUM(exp.failed_records) * 100.0 /
        NULLIF(SUM(exp.passed_records) + SUM(exp.failed_records), 0),
        2
    ) AS failure_rate_pct
FROM event_log(TABLE(my_catalog.my_schema.my_pipeline)),
     LATERAL VIEW EXPLODE(
         from_json(
             details:flow_progress.data_quality.expectations,
             'ARRAY<STRUCT<name:STRING, passed_records:BIGINT, failed_records:BIGINT>>'
         )
     ) AS exp
WHERE event_type = 'flow_progress'
GROUP BY 1, 2, 3
ORDER BY check_date DESC, failure_rate_pct DESC;