변환에 데이터 기대 조건 연결하기

데이터 expectation은 입력이나 출력이 만족해야 할 조건이고, Check는 그 조건에 이름과 실패 시 동작을 붙여 변환에 연결하는 객체입니다. Code Repositories의 Libraries에서 transforms-expectations가 설치되어 있는지 확인하고, 없으면 설치합니다.

from transforms.api import Check
from transforms import expectations as E

Check의 이름과 실패 정책

Check(expectation, name, on_error="FAIL")로 검사 하나를 만듭니다.

이름은 Data Health와 Builds 등에서 검사를 식별합니다. 단일 노드 변환의 가상 테이블 출력에서는 on_error="FAIL"을 지원하지 않습니다.

각 Check는 입력 또는 출력 하나에 연결합니다. 하나이면 checks=check, 여러 개이면 checks=[check1, check2]로 전달합니다. 조건별로 다른 정책이나 알림을 적용하려면 Check를 나눕니다. 복합 expectation 하나로 묶은 Check는 전체가 하나의 모니터링 단위입니다.

기본 키 검사: 네 엔진

다음 예제는 가상의 입력을 복사하고 출력 id가 NULL이 아니며 유일한지 검사합니다. 실제 경로로 바꿔 사용합니다.

Polars

from transforms import expectations as E
from transforms.api import Check, Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output(
        "/Example Project/Outputs/checked_polars",
        checks=Check(E.primary_key("id"), "Primary Key", 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


@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output(
        "/Example Project/Outputs/checked_duckdb",
        checks=Check(E.primary_key("id"), "Primary Key", 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


@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output(
        "/Example Project/Outputs/checked_pandas",
        checks=Check(E.primary_key("id"), "Primary Key", 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


@transform_df(
    Output(
        "/Example Project/Outputs/checked_spark",
        checks=Check(E.primary_key("id"), "Primary Key", on_error="FAIL"),
    ),
    source=Input("/Example Project/Inputs/source"),
)
def compute(source):
    return source

각 코드는 별도 변환이므로 같은 검사 이름을 사용합니다. 한 변환 안에 여러 출력을 함께 만들 때는 이름을 구분합니다.

자료형 검사 라이브러리

단일 노드 검사는 실제 계산 엔진이 Polars, Pandas 또는 DuckDB 중 무엇이든 Polars 자료형을 사용합니다. Spark 검사는 Spark SQL 자료형을 사용합니다.

import polars as pl
from transforms import expectations as E

age_type = E.col("age").has_type(pl.Int32)

from pyspark.sql import types as T
from transforms import expectations as E

age_type = E.col("age").has_type(T.LongType())

복합 조건은 동일한 나이 조건을 입력 경고와 출력 실패로 나누어 적용하는 네 엔진의 예제입니다. 열과 그룹 및 schema 참조, Timestamp 조건, 다른 데이터셋과의 비교에서 사용할 수 있는 연산과 NULL 및 시간 경계 규칙을 설명합니다.