Code Workspaces의 Jupyter에서 Iceberg 사용하기

Jupyter Code Workspace에서는 PyIceberg, SQL 또는 Spark로 Iceberg 테이블을 사용할 수 있습니다. 가상 Iceberg 테이블을 SQL로 조회하는 경로는 소스를 작업 공간에 직접 추가하거나 소스의 코드 가져오기를 켤 필요가 없습니다. 소스를 사용자 코드에 노출하지 않으려면 이 경로를 검토합니다.

Data 패널의 테이블 가져오기

Jupyter 작업 공간을 엽니다.
Data > Add > Read data에서 Iceberg 테이블을 추가합니다.

작업 공간에서 Iceberg 데이터 추가

Import table의 설치 안내로 해당 환경의 클라이언트 의존성을 준비합니다.
테이블 별칭을 지정하고 생성된 코드로 읽기 및 쓰기를 구성합니다.

transforms-tables 설치 안내와 가져오기 모드

UI의 모드 이름은 transforms-table이며 라이브러리 패키지 이름은 transforms-tables입니다. 작업 공간이 제시하는 전체 설치 명령을 사용합니다.

containers-sql 모드

같은 Read data 절차로 테이블을 추가한 뒤 Import table의 모드를 transforms-table에서 containers-sql로 바꿉니다. 해당 환경의 SQL 클라이언트 설치 및 사용 코드를 적용합니다.

containers-sql 모드 선택

가상 테이블에서 이 경로를 사용하는 경우에는 소스 코드 가져오기 권한과 별개로 SQL 조회를 구성할 수 있습니다.

수동 Spark 세션

이 경로에는 PySpark가 동작하는 Code Workspace와 다음 준비가 필요합니다.

Spark 3.5 및 Scala 2.12에 맞는 Iceberg Spark runtime JAR
같은 Iceberg 릴리스의 aws-bundle JAR
JAR을 저장할 작업 공간의 libs 디렉터리
기반 저장소 버킷의 외부 통신 정책 가져오기

예제 파일 이름은 iceberg-spark-runtime-3.5_2.12-1.9.1.jar와 iceberg-aws-bundle-1.9.1.jar입니다. 사용할 릴리스와 Spark/Scala 조합을 맞추고 실제 업로드 위치를 사용합니다. 외부 클라이언트의 형식 및 CSE 지원은 Foundry 관리 런타임의 지원과 같다고 가정하지 않습니다.

다음 함수는 실제 Foundry 기본 URL과 두 JAR의 로컬 경로를 받습니다. 새 Spark 세션을 만들기 전에 JAR과 extension을 설정합니다.

from getpass import getpass
from pathlib import Path
from pyspark.sql import SparkSession


def create_iceberg_spark(foundry_url: str, runtime_jar_path: str, aws_bundle_jar_path: str):
    jars = ",".join([
        Path(runtime_jar_path).resolve().as_uri(),
        Path(aws_bundle_jar_path).resolve().as_uri(),
    ])
    return (
        SparkSession.builder
        .master("local[*]")
        .appName("foundry")
        .config("spark.jars", jars)
        .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
        .config("spark.sql.catalog.foundry", "org.apache.iceberg.spark.SparkCatalog")
        .config("spark.sql.catalog.foundry.type", "rest")
        .config("spark.sql.catalog.foundry.uri", foundry_url.rstrip("/") + "/iceberg")
        .config("spark.sql.catalog.foundry.default-namespace", "foundry")
        .config("spark.sql.catalog.foundry.token", getpass("Foundry 토큰: "))
        .config("spark.sql.defaultCatalog", "foundry")
        .getOrCreate()
    )

카탈로그 설정	역할
`spark.sql.catalog.foundry`	Iceberg SparkCatalog 구현 클래스
`.type = rest`	REST 카탈로그 구현
`.uri`	Foundry 카탈로그 URL
`.default-namespace = foundry`	기본 이름 공간
`.token`	호출자의 실제 토큰

이미 실행 중인 Spark 세션을 재사용하면 시작 시 적용하는 JAR이나 extension 변경이 그대로 반영되지 않을 수 있습니다. 실제 세션의 구성을 확인합니다.

테이블 생성과 추가

다음 두 방식은 같은 가상 택시 데이터를 사용하는 별도 예제입니다. 실제로 새 테이블을 만들고 데이터를 쓰므로 각각 사용할 출력 경로를 정합니다. Spark에서 공백이 있는 경로는 백틱으로 인용합니다. PyIceberg의 REST 경로에는 URL 인코딩을 사용하며 논리 식별자와 이미 인코딩된 URL을 혼용하지 않습니다.

PySpark DataFrame

from pyspark.sql.types import DoubleType, FloatType, LongType, StringType, StructField, StructType


def create_taxis_with_dataframe(spark, table_path: str):
    identifier = "`" + table_path.replace("`", "``") + "`"
    schema = StructType([
        StructField("vendor_id", LongType(), True),
        StructField("trip_id", LongType(), True),
        StructField("trip_distance", FloatType(), True),
        StructField("fare_amount", DoubleType(), True),
        StructField("store_and_fwd_flag", StringType(), True),
    ])
    spark.createDataFrame([], schema).writeTo(identifier).create()
    stored_schema = spark.table(identifier).schema
    rows = [
        (1, 1000371, 1.8, 15.32, "N"),
        (2, 1000372, 2.5, 22.15, "N"),
        (2, 1000373, 0.9, 9.01, "N"),
        (1, 1000374, 8.4, 42.13, "Y"),
    ]
    spark.createDataFrame(rows, stored_schema).writeTo(identifier).append()

Spark SQL

def create_taxis_with_sql(spark, table_path: str):
    identifier = "`" + table_path.replace("`", "``") + "`"
    spark.sql(f"""
        CREATE TABLE {identifier} (
            vendor_id BIGINT,
            trip_id BIGINT,
            trip_distance FLOAT,
            fare_amount DOUBLE,
            store_and_fwd_flag STRING
        ) PARTITIONED BY (vendor_id)
    """)
    spark.sql(f"""
        INSERT INTO {identifier} VALUES
          (1, 1000371, 1.8, 15.32, 'N'),
          (2, 1000372, 2.5, 22.15, 'N'),
          (2, 1000373, 0.9, 9.01, 'N'),
          (1, 1000374, 8.4, 42.13, 'Y')
    """)

SQL 예제는 vendor_id로 파티션을 지정합니다. 수동 클라이언트 호출의 생성과 추가는 별도 동작이며 Foundry 빌드 작업 전체의 원자적 커밋으로 취급하지 않습니다. 브랜치를 명시하려면 X-PLTR-Branch 조건을 확인합니다.

Keyboard shortcuts