Iceberg 테이블 입출력 예제

네 가지 입출력 구성에 대해 Polars, DuckDB, pandas와 PySpark 예제를 제공합니다. transforms-tables 패키지와 선택한 연산 라이브러리를 준비합니다. 경로는 가상 예시이므로 실제 입력 리소스와 출력 위치로 변경하고 목적에 맞는 예제를 선택합니다.

모두 비증분 예제입니다. TableInput/TableOutput은 Iceberg 리소스이고 Input/Output은 일반 데이터셋입니다. 같은 위치를 두 리소스 유형으로 동시에 사용하지 않습니다. API와 런타임 핸들의 차이는 기본 사용법에 있습니다.

Iceberg 입력의 두 DuckDB 복사 예제는 네이티브 PyIceberg의 Arrow batch reader를 사용합니다. 이 읽기 경로는 CSE를 지원하지 않습니다. CSE가 필요한 단일 노드 읽기는 Foundry의 .arrow() 같은 API 경로를 사용하며, 이 경우 필터 푸시다운의 제약을 따릅니다. 형식 버전과 실제 클라이언트의 지원도 함께 확인합니다.

간단한 Iceberg 테이블 생성

새 테이블에 쓰는 Hello와 World는 생성용 예시 데이터입니다.

Polars

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


@transform.using(output=TableOutput("/Examples/iceberg/hello-polars"))
def create_polars(output: IcebergOutput):
    output.write_table(pl.DataFrame({"phrase": ["Hello", "World"]}))

DuckDB

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


@transform.using(output=TableOutput("/Examples/iceberg/hello-duckdb"))
def create_duckdb(ctx: LightweightContext, output: IcebergOutput):
    result = ctx.duckdb().conn.sql("SELECT * FROM (VALUES ('Hello'), ('World')) AS t(phrase)")
    output.write_table(result.to_arrow_table())

pandas

import pandas as pd
from transforms.api import transform
from transforms.tables import IcebergOutput, TableOutput


@transform.using(output=TableOutput("/Examples/iceberg/hello-pandas"))
def create_pandas(output: IcebergOutput):
    output.write_table(pd.DataFrame({"phrase": ["Hello", "World"]}))

PySpark

from transforms.api import TransformContext, transform
from transforms.tables import TableOutput, TableTransformOutput


@transform(output=TableOutput("/Examples/iceberg/hello-spark"))
def create_spark(ctx: TransformContext, output: TableTransformOutput):
    frame = ctx.spark_session.createDataFrame([["Hello"], ["World"]], schema=["phrase"])
    output.write_dataframe(frame)

Iceberg 입력과 Iceberg 출력

Polars

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


@transform.using(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_table=TableOutput("/Examples/iceberg/copied-polars"),
)
def copy_table_polars(source_table: IcebergInput, output_table: IcebergOutput):
    output_table.write_table(source_table.polars())

DuckDB

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


@transform.using(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_table=TableOutput("/Examples/iceberg/copied-duckdb"),
)
def copy_table_duckdb(ctx: LightweightContext, source_table: IcebergInput, output_table: IcebergOutput):
    reader = source_table.table().scan().to_arrow_batch_reader()
    connection = ctx.duckdb().conn
    connection.register("source_rows", reader)
    result = connection.sql("SELECT * FROM source_rows")
    output_table.write_table(result.to_arrow_table())

pandas

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


@transform.using(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_table=TableOutput("/Examples/iceberg/copied-pandas"),
)
def copy_table_pandas(source_table: IcebergInput, output_table: IcebergOutput):
    output_table.write_table(source_table.pandas())

PySpark

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


@transform(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_table=TableOutput("/Examples/iceberg/copied-spark"),
)
def copy_table_spark(source_table: TableTransformInput, output_table: TableTransformOutput):
    output_table.write_dataframe(source_table.dataframe())

데이터셋 입력과 Iceberg 출력

Polars

from transforms.api import Input, LightweightInput, transform
from transforms.tables import IcebergOutput, TableOutput


@transform.using(
    source_dataset=Input("/Examples/iceberg/input-dataset"),
    output_table=TableOutput("/Examples/iceberg/from-dataset-polars"),
)
def dataset_to_table_polars(source_dataset: LightweightInput, output_table: IcebergOutput):
    output_table.write_table(source_dataset.polars())

DuckDB

from transforms.api import Input, LightweightContext, LightweightInput, transform
from transforms.tables import IcebergOutput, TableOutput


@transform.using(
    source_dataset=Input("/Examples/iceberg/input-dataset"),
    output_table=TableOutput("/Examples/iceberg/from-dataset-duckdb"),
)
def dataset_to_table_duckdb(ctx: LightweightContext, source_dataset: LightweightInput, output_table: IcebergOutput):
    connection = ctx.duckdb().conn
    connection.register("source_rows", source_dataset.arrow())
    result = connection.sql("SELECT * FROM source_rows")
    output_table.write_table(result.to_arrow_table())

pandas

from transforms.api import Input, LightweightInput, transform
from transforms.tables import IcebergOutput, TableOutput


@transform.using(
    source_dataset=Input("/Examples/iceberg/input-dataset"),
    output_table=TableOutput("/Examples/iceberg/from-dataset-pandas"),
)
def dataset_to_table_pandas(source_dataset: LightweightInput, output_table: IcebergOutput):
    output_table.write_table(source_dataset.pandas())

PySpark

from transforms.api import Input, TransformInput, transform
from transforms.tables import TableOutput, TableTransformOutput


@transform(
    source_dataset=Input("/Examples/iceberg/input-dataset"),
    output_table=TableOutput("/Examples/iceberg/from-dataset-spark"),
)
def dataset_to_table_spark(source_dataset: TransformInput, output_table: TableTransformOutput):
    output_table.write_dataframe(source_dataset.dataframe())

Iceberg 입력과 데이터셋 출력

Polars

from transforms.api import LightweightOutput, Output, transform
from transforms.tables import IcebergInput, TableInput


@transform.using(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_dataset=Output("/Examples/iceberg/as-dataset-polars"),
)
def table_to_dataset_polars(source_table: IcebergInput, output_dataset: LightweightOutput):
    output_dataset.write_table(source_table.polars())

DuckDB

from transforms.api import LightweightContext, LightweightOutput, Output, transform
from transforms.tables import IcebergInput, TableInput


@transform.using(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_dataset=Output("/Examples/iceberg/as-dataset-duckdb"),
)
def table_to_dataset_duckdb(ctx: LightweightContext, source_table: IcebergInput, output_dataset: LightweightOutput):
    reader = source_table.table().scan().to_arrow_batch_reader()
    connection = ctx.duckdb().conn
    connection.register("source_rows", reader)
    result = connection.sql("SELECT * FROM source_rows")
    output_dataset.write_table(result)

pandas

from transforms.api import LightweightOutput, Output, transform
from transforms.tables import IcebergInput, TableInput


@transform.using(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_dataset=Output("/Examples/iceberg/as-dataset-pandas"),
)
def table_to_dataset_pandas(source_table: IcebergInput, output_dataset: LightweightOutput):
    output_dataset.write_table(source_table.pandas())

PySpark

from transforms.api import Output, TransformOutput, transform
from transforms.tables import TableInput, TableTransformInput


@transform(
    source_table=TableInput("/Examples/iceberg/input-table"),
    output_dataset=Output("/Examples/iceberg/as-dataset-spark"),
)
def table_to_dataset_spark(source_table: TableTransformInput, output_dataset: TransformOutput):
    output_dataset.write_dataframe(source_table.dataframe())

DuckDB는 실제 Arrow 데이터 또는 batch reader를 SQL 이름으로 등록한 뒤 질의합니다. 데이터셋 런타임 핸들 자체를 관계형 테이블로 간주하지 않습니다. Iceberg 출력에는 Arrow Table을 전달하고, 일반 LightweightOutput 예제는 DuckDB relation을 직접 기록합니다.

Keyboard shortcuts

팔란티어 플랫폼 한국어 해설