Pipeline 등록과 변환 생성

Python transforms 하위 프로젝트는 transforms.api.Pipeline 객체 하나를 노출합니다. Pipeline은 출력 데이터셋의 빌드 방법을 등록하고, 실제 빌드 때 해당 계산 함수를 찾는 역할을 합니다.

기본 템플릿은 자동 등록을 구성하므로 보통 datasets 패키지에 변환을 추가하면 됩니다.

자동 등록

src/myproject/pipeline.py의 일반적인 구성은 다음과 같습니다.

from transforms.api import Pipeline
from myproject import datasets

my_pipeline = Pipeline()
my_pipeline.discover_transforms(datasets)

discover_transforms()는 지정한 모듈이나 패키지 안의 모듈을 재귀적으로 import하여 모듈 수준의 변환 정의를 찾습니다. 이 과정에서 import한 모듈의 전역 코드도 실행됩니다. 변환 선언을 넘어 실제 데이터 처리나 외부 부작용을 import 시점에 실행하지 않도록 구분해야 합니다.

수동 등록

등록할 변환을 명시적으로 제어하려면 add_transforms()를 사용합니다. 한 번에 여러 변환을 받을 수 있으며 서로 다른 변환이 같은 출력 자산을 선언하는지도 검사합니다.

import polars as pl
from transforms.api import Input, Output, Pipeline, transform


@transform.using(
    my_output=Output("/Example Project/Outputs/adults"),
    my_input=Input("/Example Project/Inputs/people"),
)
def select_adults(my_output, my_input):
    my_output.write_table(my_input.polars().filter(pl.col("age") >= 18))


my_pipeline = Pipeline()
my_pipeline.add_transforms(select_adults)

이 예시는 age 열이 있는 실제 입력을 요구합니다. 같은 Pipeline을 다른 모듈에 둔다면 그 변환을 import하여 기존 Pipeline에 추가합니다.

반복되는 변환 생성

같은 로직을 여러 지역이나 여러 원천에 적용하려면 변환 객체들을 생성하여 등록할 수 있습니다. 아래 세 예제는 각각 하나의 엔진을 선택하는 대안이며, 한 저장소에서 같은 출력 경로의 세 구현을 동시에 등록하지 않습니다.

공통 입력 계약은 src1, src2, src3별 입력 데이터셋에 String형 source 열이 있다는 것입니다.

Polars

import polars as pl
from transforms.api import Input, Output, transform


def make_transform(source_name):
    @transform.using(
        my_input=Input(f"/Example Project/Inputs/{source_name}"),
        output=Output(f"/Example Project/Outputs/{source_name}"),
    )
    def compute(my_input, output):
        rows = my_input.polars().filter(pl.col("source") == source_name)
        output.write_table(rows)

    compute.__name__ = f"filter_{source_name}"
    return compute


def transform_generator(sources):
    return [make_transform(source) for source in sources]


TRANSFORMS = transform_generator(["src1", "src2", "src3"])

DuckDB

from transforms.api import Input, Output, transform


def make_transform(source_name):
    @transform.using(
        my_input=Input(f"/Example Project/Inputs/{source_name}"),
        output=Output(f"/Example Project/Outputs/{source_name}"),
    )
    def compute(ctx, my_input, output):
        query = ctx.duckdb().conn.sql(
            "SELECT * FROM my_input WHERE source = ?",
            params=[source_name],
        )
        output.write_table(query)

    compute.__name__ = f"filter_{source_name}"
    return compute


def transform_generator(sources):
    return [make_transform(source) for source in sources]


TRANSFORMS = transform_generator(["src1", "src2", "src3"])

값은 SQL 문자열에 직접 삽입하지 않고 매개변수로 전달합니다. 데이터셋은 선언한 입력 이름인 my_input으로 등록됩니다.

Pandas

from transforms.api import Input, Output, transform


def make_transform(source_name):
    @transform.using(
        my_input=Input(f"/Example Project/Inputs/{source_name}"),
        output=Output(f"/Example Project/Outputs/{source_name}"),
    )
    def compute(my_input, output):
        rows = my_input.pandas()
        output.write_table(rows.loc[rows["source"] == source_name])

    compute.__name__ = f"filter_{source_name}"
    return compute


def transform_generator(sources):
    return [make_transform(source) for source in sources]


TRANSFORMS = transform_generator(["src1", "src2", "src3"])

각 factory 호출은 별도 source_name을 보관합니다. 같은 반복문 변수를 나중에 읽는 closure가 모두 마지막 값으로 바뀌는 문제를 피합니다. 기본값 인자에 현재 값을 고정하는 방식도 가능하지만, 위 구성은 계산 함수에 선언되지 않은 자산 인자를 추가하지 않습니다.

생성 결과 등록

생성한 변환들은 모듈 수준의 리스트 변수에 보관해야 자동 발견 경로에서 모두 찾을 수 있습니다. 루프가 끝난 뒤 마지막 함수 변수만 남겨 두는 대신 위 예제처럼 TRANSFORMS 전체를 유지합니다.

파일을 myproject/datasets/generated.py에 두었다면 수동으로도 추가할 수 있습니다.

from transforms.api import Pipeline
from myproject.datasets import generated

my_pipeline = Pipeline()
my_pipeline.add_transforms(*generated.TRANSFORMS)

여러 변환을 생성하면 출력마다 입력을 다시 읽거나 계산할 수 있습니다. 입력과 공통 로직을 한 번 처리하여 여러 출력을 쓰는 것이 더 적합한 경우에는 다중 출력 변환을 검토합니다.

CI에서 확정하는 입출력

CI가 실행될 때 모든 입출력 관계와 데이터셋 이름 및 RID 연결을 결정합니다. 없는 출력 자산을 만들고 JobSpec에 저장소, 소스 파일과 계산 함수의 entry point를 기록합니다.

빌드 시점에 입력 목록을 읽어 다른 자산으로 바꾸거나 출력 이름을 동적으로 바꾸는 방식은 지원하지 않습니다. CI 이후 구성 파일의 입력 목록이 달라지면 새 참조가 JobSpec에 없어 빌드가 실패할 수 있습니다. 입출력 변경은 commit과 checks를 통해 새 정의로 발행해야 합니다.

수동 등록의 경우 Code Repositories의 파일별 Build 버튼이 No transforms discovered in the pipeline from the requested file 오류를 보일 수 있습니다. 등록한 데이터셋은 Data Lineage나 Dataset Preview에서 빌드할 수 있습니다.

Entry point

기본 구조를 바꾸거나 여러 파이프라인을 포함한 저장소를 구성할 때는 setup.py에서 실제 Pipeline 위치를 맞춥니다. 기본 계약은 각 Python 하위 프로젝트가 transforms.pipelines 그룹에 root라는 entry point를 하나 노출하는 것입니다.

아래 값은 모듈 myproject.pipeline 안의 my_pipeline 객체를 가리킵니다.

entry_points = {
    "transforms.pipelines": ["root = myproject.pipeline:my_pipeline"]
}

root는 entry point의 이름이고, my_pipeline은 Python 객체의 속성 이름입니다. 기존 setup()의 다른 메타데이터를 보존하면서 entry_points 인자에 이 매핑을 전달합니다. 전체 파일 구성은 프로젝트 구조를 따릅니다.