> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sifi.life/llms.txt
> Use this file to discover all available pages before exploring further.

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

> 이 문서는 **[데이터 엔지니어링](/training/05-data-engineering/overview)** 섹션의 일부입니다.

## 왜 데이터 품질 관리가 필요한가?

데이터 파이프라인에서 가장 흔한 문제는 \*\*"Garbage In, Garbage Out"\*\*입니다. 소스 데이터에 결측값, 잘못된 형식, 비즈니스 규칙 위반 등이 포함되면, 이를 기반으로 만들어진 리포트와 ML 모델의 결과도 신뢰할 수 없게 됩니다.

> 💡 전통적으로 데이터 품질 검증은 파이프라인 코드 곳곳에 `if` 문이나 `filter()` 로직을 수동으로 삽입해야 했습니다. 이 방식은 유지보수가 어렵고, 품질 규칙의 전체 현황을 한눈에 파악하기 어렵습니다. SDP의 **Expectations** 는 이 문제를 **선언적(Declarative)** 으로 해결합니다.

| 검증 결과 | 동작              | 설명                      |
| ----- | --------------- | ----------------------- |
| 통과    | 정상 데이터 → 다음 단계  | 데이터가 품질 기준을 충족합니다       |
| WARN  | 경고 기록 + 데이터 포함  | 위반 사항을 기록하되 데이터는 통과시킵니다 |
| DROP  | 위반 행 제거 (격리 가능) | 위반된 행을 제거합니다            |
| FAIL  | 파이프라인 중지        | 파이프라인 실행을 즉시 중지합니다      |

***

## Expectations란?

**Expectations** 는 SDP(Spark Declarative Pipelines)에서 데이터 품질 규칙을 **선언적으로** 정의하는 기능입니다. SQL의 `CONSTRAINT`처럼 테이블 정의에 품질 규칙을 직접 명시하며, 각 규칙에 대해 부적합 데이터를 어떻게 처리할지(경고, 삭제, 파이프라인 중지)를 지정할 수 있습니다.

### 핵심 특징

| 특징           | 설명                                                |
| ------------ | ------------------------------------------------- |
| **선언적 정의**   | 테이블 DDL 안에 `CONSTRAINT ... EXPECT (조건)` 형태로 작성합니다 |
| **자동 모니터링**  | 매 업데이트마다 통과/위반 건수가 자동으로 기록됩니다                     |
| **유연한 처리**   | 규칙별로 WARN, DROP ROW, FAIL UPDATE 중 선택할 수 있습니다     |
| **파이프라인 통합** | 별도 도구 없이 SDP 파이프라인 안에서 바로 사용할 수 있습니다              |

***

## 위반 처리 방식 3가지

Expectations는 데이터 품질 위반 시 **세 가지 행동** 을 지원합니다. 비즈니스 요구사항에 따라 규칙별로 다르게 설정할 수 있습니다.

### 상세 비교

| 방식           | SQL 키워드                    | Python 데코레이터                               | 위반 시 동작                         | 사용 시나리오                       |
| ------------ | -------------------------- | ------------------------------------------ | ------------------------------- | ----------------------------- |
| **경고만**      | `ON VIOLATION WARN` (기본)   | `@dlt.expect("name", "condition")`         | 위반 레코드를 **포함** 하되, 위반 건수를 기록합니다 | 모니터링 목적, 신규 규칙 테스트            |
| **행 삭제**     | `ON VIOLATION DROP ROW`    | `@dlt.expect_or_drop("name", "condition")` | 위반 레코드를 결과에서 **제외** 합니다         | 데이터 품질 보장이 필요한 Silver/Gold 계층 |
| **파이프라인 중지** | `ON VIOLATION FAIL UPDATE` | `@dlt.expect_or_fail("name", "condition")` | 위반 발생 시 파이프라인을 **즉시 중지** 합니다    | 치명적 오류(스키마 위반, 필수값 누락 등)      |

> ⚠️ **FAIL UPDATE 사용 시 주의**: 파이프라인이 중지되면 해당 업데이트의 **모든** 테이블 갱신이 중단됩니다. 복구하려면 문제 데이터를 수정한 후 파이프라인을 다시 실행해야 합니다.

### 동작 흐름 비교

**위반 처리 모드별 동작 비교**

| 모드              | 통과 시    | 위반 시                  |
| --------------- | ------- | --------------------- |
| **WARN**        | 테이블에 저장 | 테이블에 저장 + 위반 메트릭 기록   |
| **DROP ROW**    | 테이블에 저장 | 레코드 제거 + 위반 메트릭 기록    |
| **FAIL UPDATE** | 테이블에 저장 | 파이프라인 즉시 중지 + 업데이트 실패 |

***

## Expectation 정의 문법

### SQL 문법

SQL에서는 `CREATE OR REFRESH STREAMING TABLE` 또는 `CREATE OR REFRESH MATERIALIZED VIEW` 정의 안에 `CONSTRAINT` 절로 Expectation을 선언합니다.

```sql theme={null}
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);
```

> 💡 `ON VIOLATION WARN`은 기본값이므로, 생략하면 자동으로 WARN이 적용됩니다. 명시적으로 작성하는 것이 가독성에 좋습니다.

### Python 문법

Python에서는 `@dlt.expect`, `@dlt.expect_or_drop`, `@dlt.expect_or_fail` 데코레이터를 사용합니다.

```python theme={null}
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")
```

***