증분 변환에서 전체 참조 표 읽기

국가 코드처럼 주기적으로 전체를 다시 가져오는 참조 표는 새 입력 행을 해석할 때마다 현재 전체 내용을 읽어야 할 수 있습니다. 이 입력을 snapshot_inputs에 등록하면 입력이 전체 교체되거나 파일을 갱신/삭제해도 그 사실만으로 변환의 증분 실행을 막지 않습니다.

목록에는 데이터셋 경로나 RID가 아니라 decorator의 입력 인자 이름을 적습니다. 스냅샷 참조 입력의 added와 current는 증분/비증분 실행 모두에서 전체 입력을 반환하고, 다른 읽기 모드는 빈 입력을 반환합니다. 이전에 본 버전과 시작 트랜잭션이 달라도 허용합니다.

국가 코드 조인

다음 예제에서 phone_numbers는 새 전화번호를 추가하는 데이터셋이며 country_code 열이 있습니다. country_codes는 code 열로 국가 정보를 찾는 전체 참조 표입니다. 가상의 경로를 실제 입력으로 바꿉니다.

참조 표의 변경이 이미 계산한 결과를 무효화하지 않는 경우에 이 패턴을 사용합니다. 참조 값을 바꾼 뒤 과거 출력까지 새 값으로 고쳐야 한다면 새 행에만 조인해서는 부족하므로 전체 재계산을 검토합니다. 참조 키가 중복되어 있으면 한 입력 행이 여러 결과와 일치할 수도 있습니다.

Polars

from transforms.api import Input, Output, incremental, transform


@incremental(snapshot_inputs=["country_codes"])
@transform.using(
    phone_numbers=Input("/Example Project/Inputs/phone_numbers"),
    country_codes=Input("/Example Project/Inputs/country_codes"),
    output=Output("/Example Project/Outputs/phone_countries_polars"),
)
def compute(phone_numbers, country_codes, output):
    phones = phone_numbers.polars()
    countries = country_codes.polars()
    output.write_table(
        phones.join(countries, left_on="country_code", right_on="code", how="left")
    )

DuckDB

from transforms.api import Input, Output, incremental, transform


@incremental(snapshot_inputs=["country_codes"])
@transform.using(
    phone_numbers=Input("/Example Project/Inputs/phone_numbers"),
    country_codes=Input("/Example Project/Inputs/country_codes"),
    output=Output("/Example Project/Outputs/phone_countries_duckdb"),
)
def compute(ctx, phone_numbers, country_codes, output):
    connection = ctx.duckdb().conn
    result = connection.sql("""
        SELECT *
        FROM phone_numbers
        LEFT JOIN country_codes
          ON phone_numbers.country_code = country_codes.code
    """)
    output.write_table(result)

Pandas

from transforms.api import Input, Output, incremental, transform


@incremental(snapshot_inputs=["country_codes"])
@transform.using(
    phone_numbers=Input("/Example Project/Inputs/phone_numbers"),
    country_codes=Input("/Example Project/Inputs/country_codes"),
    output=Output("/Example Project/Outputs/phone_countries_pandas"),
)
def compute(phone_numbers, country_codes, output):
    phones = phone_numbers.pandas()
    countries = country_codes.pandas()
    output.write_table(
        phones.merge(countries, left_on="country_code", right_on="code", how="left")
    )

PySpark

from transforms.api import Input, Output, incremental, transform


@incremental(snapshot_inputs=["country_codes"])
@transform.spark.using(
    phone_numbers=Input("/Example Project/Inputs/phone_numbers"),
    country_codes=Input("/Example Project/Inputs/country_codes"),
    output=Output("/Example Project/Outputs/phone_countries_spark"),
)
def compute(phone_numbers, country_codes, output):
    phones = phone_numbers.dataframe()
    countries = country_codes.dataframe()
    output.write_dataframe(
        phones.join(countries, phones.country_code == countries.code, "left_outer")
    )

첫 실행에서는 전화번호 전체를 처리하여 출력을 만들고, 이후에는 새 전화번호에 현재 참조 표를 조인합니다. 엔진마다 NULL 조인 키를 처리하는 방식이 다를 수 있으므로 NULL이 가능한 입력은 해당 엔진의 조인 의미에 맞춰 설계합니다. 예를 들어 Pandas의 기본 merge는 양쪽의 NULL 키를 일치시키지만, 위 DuckDB와 Spark의 일반 등가 조인은 NULL 키를 일치시키지 않습니다.

Cipher 자산과 입력 구성 변경

bellaso_python_lib로 Cipher 자산을 사용하는 증분 변환에서는 encrypter, decrypter와 hasher도 snapshot_inputs에 포함해야 합니다. 해당 자산을 읽는 입력 인자 이름을 목록에 지정합니다.

스냅샷 참조 입력은 이전 버전의 연속성을 검사하지 않으므로 추가하거나 제거하면서 증분 실행을 유지할 수 있습니다. 입력 구성 변경으로 기존 출력의 의미가 바뀌면 semantic version을 바꿔 재계산할지 판단합니다.

Keyboard shortcuts