Python 변환의 Iceberg 입출력

Iceberg 변환은 데코레이터의 입력 및 출력 선언, 실행 함수가 받는 런타임 핸들, 그리고 실제 연산으로 구성됩니다. transforms-tables 패키지가 제공하는 transforms.tables를 사용합니다.

Iceberg 변환의 선언, 런타임 인수와 연산

선언과 런타임 구분

선언	의미	일반 데이터셋의 대응
`TableInput`	Iceberg 입력 리소스를 선언합니다.	`transforms.api.Input`
`TableOutput`	Iceberg 출력 리소스를 선언합니다.	`transforms.api.Output`

단일 노드는 @transform.using, PySpark는 @transform.spark.using을 사용할 수 있습니다. 기존 Spark @transform 예제도 같은 테이블 선언을 사용합니다.

역할	단일 노드 핸들	PySpark 핸들
일반 입력	`IcebergInput`	`TableTransformInput`
일반 출력	`IcebergOutput`	`TableTransformOutput`
증분 입력	`IncrementalIcebergInput`	`IncrementalTableTransformInput`
증분 출력	`IncrementalIcebergOutput`	`TableTransformOutput`

일반 데이터셋에서는 각각 LightweightInput/TransformInput, LightweightOutput/TransformOutput 및 증분 입력/출력 핸들을 사용합니다. 같은 이름의 메서드가 있더라도 반환 유형을 확인합니다.

읽기와 쓰기

실행 경로	읽기	반환 유형	쓰기
단일 노드 Polars	`.polars()`	Polars LazyFrame	`.write_table()`
단일 노드 pandas	`.pandas()`	pandas DataFrame	`.write_table()`
단일 노드 Arrow	`.arrow()`	PyArrow Table	`.write_table()`
PySpark	`.dataframe()`	PySpark DataFrame	`.write_dataframe()`

읽은 뒤의 필터나 집계는 해당 DataFrame 라이브러리의 일반 API를 사용합니다. Iceberg를 읽고 쓰는 부분은 런타임 핸들이 처리합니다. Arrow는 DuckDB에 등록하여 SQL로 처리할 수도 있습니다.

아래 예제의 입력에는 문자열 열 status가 있다고 가정합니다. 경로는 가상 위치이며 실제 테이블로 바꿉니다.

Polars

import polars as pl
from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    source=TableInput("/Examples/iceberg/source"),
    output=TableOutput("/Examples/iceberg/active-polars"),
)
def filter_polars(source: IcebergInput, output: IcebergOutput):
    active = source.polars().filter(pl.col("status") == "active")
    output.write_table(active)

DuckDB

from transforms.api import LightweightContext, transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    source=TableInput("/Examples/iceberg/source"),
    output=TableOutput("/Examples/iceberg/active-duckdb"),
)
def filter_duckdb(ctx: LightweightContext, source: IcebergInput, output: IcebergOutput):
    connection = ctx.duckdb().conn
    connection.register("source_rows", source.arrow())
    result = connection.sql("SELECT * FROM source_rows WHERE status = 'active'")
    output.write_table(result.to_arrow_table())

pandas

from transforms.api import transform
from transforms.tables import IcebergInput, IcebergOutput, TableInput, TableOutput


@transform.using(
    source=TableInput("/Examples/iceberg/source"),
    output=TableOutput("/Examples/iceberg/active-pandas"),
)
def filter_pandas(source: IcebergInput, output: IcebergOutput):
    frame = source.pandas()
    output.write_table(frame[frame["status"] == "active"])

PySpark

from transforms.api import transform
from transforms.tables import TableInput, TableOutput, TableTransformInput, TableTransformOutput


@transform(
    source=TableInput("/Examples/iceberg/source"),
    output=TableOutput("/Examples/iceberg/active-spark"),
)
def filter_spark(source: TableTransformInput, output: TableTransformOutput):
    frame = source.dataframe()
    output.write_dataframe(frame.filter(frame.status == "active"))

전체 입출력 조합은 16개 코드 예제에 있습니다. 행 편집, Spark 프로시저와 메타데이터를 직접 다루려면 네이티브 Iceberg 핸들을 사용합니다. 증분 실행은 별도의 데코레이터와 읽기 모드가 필요합니다.

Keyboard shortcuts