같은 기대 조건에 다른 실패 정책 적용하기

나이의 유효 범위를 0 <= age < 200으로 정의하고 NULL을 허용하지 않는 가상의 규칙을 사용합니다. 입력에서는 잘못된 나이에 경고를 남기고, 출력에서는 같은 조건과 기본 키 조건을 실패 정책으로 적용합니다. 각 Check 이름은 입력과 출력 전체에서 다릅니다.

재사용할 조건

src/myproject/datasets/shared_expectations.py에 조건을 정의합니다.

from transforms import expectations as E

VALID_AGE = E.all(
    E.col("age").non_null(),
    E.col("age").gte(0),
    E.col("age").lt(200),
)

수치 비교만으로는 NULL을 배제하지 않으므로 non_null()을 별도로 결합합니다. 같은 expression을 재사용하더라도 Check마다 이름과 실패 정책을 따로 정할 수 있습니다.

Polars

from transforms import expectations as E
from transforms.api import Check, Input, Output, transform
from myproject.datasets.shared_expectations import VALID_AGE


@transform.using(
    source=Input(
        "/Example Project/Inputs/people",
        checks=Check(VALID_AGE, "Valid age on input", on_error="WARN"),
    ),
    output=Output(
        "/Example Project/Outputs/validated_polars",
        checks=[
            Check(E.primary_key("id"), "Primary Key", on_error="FAIL"),
            Check(VALID_AGE, "Valid age on output", on_error="FAIL"),
        ],
    ),
)
def compute(source, output):
    output.write_table(source.polars())

DuckDB

from transforms import expectations as E
from transforms.api import Check, Input, Output, transform
from myproject.datasets.shared_expectations import VALID_AGE


@transform.using(
    source=Input(
        "/Example Project/Inputs/people",
        checks=Check(VALID_AGE, "Valid age on input", on_error="WARN"),
    ),
    output=Output(
        "/Example Project/Outputs/validated_duckdb",
        checks=[
            Check(E.primary_key("id"), "Primary Key", on_error="FAIL"),
            Check(VALID_AGE, "Valid age on output", on_error="FAIL"),
        ],
    ),
)
def compute(ctx, source, output):
    output.write_table(ctx.duckdb().conn.sql("SELECT * FROM source"))

Pandas

from transforms import expectations as E
from transforms.api import Check, Input, Output, transform
from myproject.datasets.shared_expectations import VALID_AGE


@transform.using(
    source=Input(
        "/Example Project/Inputs/people",
        checks=Check(VALID_AGE, "Valid age on input", on_error="WARN"),
    ),
    output=Output(
        "/Example Project/Outputs/validated_pandas",
        checks=[
            Check(E.primary_key("id"), "Primary Key", on_error="FAIL"),
            Check(VALID_AGE, "Valid age on output", on_error="FAIL"),
        ],
    ),
)
def compute(source, output):
    output.write_table(source.pandas())

PySpark

from transforms import expectations as E
from transforms.api import Check, Input, Output, transform_df
from myproject.datasets.shared_expectations import VALID_AGE


@transform_df(
    Output(
        "/Example Project/Outputs/validated_spark",
        checks=[
            Check(E.primary_key("id"), "Primary Key", on_error="FAIL"),
            Check(VALID_AGE, "Valid age on output", on_error="FAIL"),
        ],
    ),
    source=Input(
        "/Example Project/Inputs/people",
        checks=Check(VALID_AGE, "Valid age on input", on_error="WARN"),
    ),
)
def compute(source):
    return source

이 예제들은 입력을 그대로 복사하므로 입력 경고의 원인이 되는 행이 출력에도 남아 있으면 출력 검사는 실패합니다. WARN이 행을 제거하거나 값을 보정하는 기능은 아닙니다.

복합 조건의 어느 부분이 실패했는지 각각 구독하고 알림을 받으려면 나이의 NULL 여부와 범위 등을 별도 Check로 나눕니다. 하나의 Check에 결합된 expression은 전체가 하나의 모니터링 단위입니다.