아이스버그 자바 API

아이스버그 자바 API (Iceberg Java API)

아이스버그 API의 주요 목적은 스키마, 파티션 스펙, 메타데이터, 데이터 파일 같은 테이블 메타데이터를 관리하는 것이에요. 이 문서에서는 아이스버그 자바 API의 핵심 개념인 테이블, 스캔, 업데이트 연산, 트랜잭션, 타입, 표현식(expressions), 그리고 라이브러리 모듈 구조를 알려드릴게요. 테이블을 읽고 수정하는 API 사용법을 실제 코드 예시와 함께 살펴볼게요.

출처: 문서

본문

테이블 (Tables)

아이스버그 API의 주요 목적은 스키마, 파티션 스펙, 메타데이터, 그리고 테이블 데이터를 저장하는 데이터 파일 같은 테이블 메타데이터를 관리하는 것이에요.

테이블 메타데이터와 연산은 Table 인터페이스를 통해 접근해요. 이 인터페이스는 테이블 정보를 반환해요.

테이블 메타데이터 (Table metadata)

Table 인터페이스는 테이블 메타데이터에 대한 접근을 제공해요.

• schema: 현재 테이블 스키마 반환
• spec: 현재 테이블 파티션 스펙 반환
• properties: 키-값 속성 맵 반환
• currentSnapshot: 현재 테이블 스냅샷 반환
• snapshots: 테이블의 모든 유효한 스냅샷 반환
• snapshot(id): ID로 특정 스냅샷 반환
• location: 테이블의 기본 위치 반환

테이블은 또한 테이블을 최신 버전으로 갱신하는 refresh를 제공하고, 헬퍼를 노출해요.

• io: 테이블 파일을 읽고 쓰는 데 사용되는 FileIO 반환
• locationProvider: 데이터와 메타데이터 파일의 경로를 만드는 데 사용되는 LocationProvider 반환

스캔 (Scanning)

파일 수준 (File level)

아이스버그 테이블 스캔은 newScan으로 TableScan 객체를 만들면서 시작해요.

TableScan scan = table.newScan();

스캔을 구성하려면 TableScan에 filter와 select를 호출해서 그 변경 사항이 적용된 새 TableScan을 얻어요.

TableScan filteredScan = scan.filter(Expressions.equal("id", 5))

구성 메서드 호출은 새 TableScan을 만들어서, 각 TableScan이 불변(immutable)이고 스레드 간 공유돼도 예기치 않게 변하지 않도록 해요.

스캔이 구성되면 planFiles, planTasks, schema를 사용해서 파일, 작업, 읽기 프로젝션을 반환해요.

TableScan scan = table.newScan()
    .filter(Expressions.equal("id", 5))
    .select("id", "data");

Schema projection = scan.schema();
Iterable<CombinedScanTask> tasks = scan.planTasks();

타임 트래블 쿼리를 위해 asOfTime 또는 useSnapshot을 사용해서 테이블 스냅샷을 구성해요.

행 수준 (Row level)

아이스버그 테이블 스캔은 IcebergGenerics.read로 ScanBuilder 객체를 만들면서 시작해요.

ScanBuilder scanBuilder = IcebergGenerics.read(table)

스캔을 구성하려면 ScanBuilder에 where와 select를 호출해서 그 변경 사항이 적용된 새 ScanBuilder를 얻어요.

scanBuilder.where(Expressions.equal("id", 5))

스캔이 구성되면 build 메서드를 호출해서 스캔을 실행해요. build는 CloseableIterable를 반환해요.

CloseableIterable<Record> result = IcebergGenerics.read(table)
        .where(Expressions.lessThan("id", 5))
        .build();

여기서 Record는 iceberg-data 모듈의 아이스버그 레코드로, org.apache.iceberg.data.Record예요.

업데이트 연산 (Update operations)

Table은 또한 테이블을 업데이트하는 연산을 노출해요. 이 연산들은 PendingUpdate라는 빌더 패턴을 사용하고, PendingUpdate#commit이 호출될 때 커밋돼요.

예를 들어 테이블 스키마를 업데이트하는 것은 updateSchema를 호출하고, 빌더에 업데이트를 추가하고, 마지막으로 commit을 호출해서 보류 중인 변경을 테이블에 커밋하는 방식으로 수행돼요.

table.updateSchema()
    .addColumn("count", Types.LongType.get())
    .commit();

테이블을 업데이트하는 데 사용 가능한 연산은 다음과 같아요.

• updateSchema — 테이블 스키마 업데이트
• updateSpec — 테이블의 파티션 스펙 수정
• updateStatistics — 테이블의 통계 파일 업데이트
• updatePartitionStatistics — 테이블의 특정 파티션에 대한 통계 업데이트
• updateProperties — 테이블 속성 업데이트
• updateLocation — 테이블의 기본 위치 업데이트
• expireSnapshots — 테이블에서 오래된 스냅샷을 제거하는 데 사용
• manageSnapshots — 테이블 스냅샷을 관리하는 데 사용
• newAppend — 데이터 파일을 추가하는 데 사용
• newFastAppend — 데이터 파일을 추가하는 데 사용, 메타데이터를 컴팩션하지 않음
• newOverwrite — 데이터 파일을 추가하고 덮어쓰여지는 파일을 제거하는 데 사용
• newDelete — 데이터 파일을 삭제하는 데 사용
• newRewrite — 데이터 파일을 재작성하는 데 사용; 기존 파일을 새 버전으로 교체
• newRowDelta — 기존 데이터 파일에서 행을 제거하거나 교체하는 데 사용
• newTransaction — 새 테이블 수준 트랜잭션 생성
• rewriteManifests — 파일을 클러스터링해서 매니페스트 데이터 재작성, 더 빠른 스캔 계획을 위해
• replaceSortOrder — 테이블 정렬 순서를 새로 만든 순서로 교체하기 위해
• newReplacePartitions — 테이블의 파티션을 새 데이터로 동적으로 덮어쓰는 데 사용

트랜잭션 (Transactions)

트랜잭션은 여러 테이블 변경을 하나의 원자적 연산으로 커밋하는 데 사용돼요. 트랜잭션은 Table로 작업하는 것처럼 newAppend 같은 팩토리 메서드를 사용해 개별 연산을 만드는 데 사용돼요. 트랜잭션이 만든 연산들은 commitTransaction이 호출될 때 그룹으로 커밋돼요.

예를 들어 같은 트랜잭션에서 파일을 삭제하고 추가하기:

Transaction t = table.newTransaction();

// commit operations to the transaction
t.newDelete().deleteFromRowFilter(filter).commit();
t.newAppend().appendFile(data).commit();

// commit all the changes to the table
t.commitTransaction();

타입 (Types)

아이스버그 데이터 타입은 org.apache.iceberg.types 패키지에 있어요.

기본 타입 (Primitives)

기본 타입 인스턴스는 각 타입 클래스의 정적 메서드로 사용할 수 있어요. 파라미터가 없는 타입은 get을 사용하고, decimal 같은 타입은 팩토리 메서드를 사용해요.

Types.IntegerType.get()    // int
Types.DoubleType.get()     // double
Types.DecimalType.of(9, 2) // decimal(9, 2)

중첩 타입 (Nested types)

Struct, map, list는 타입 클래스의 팩토리 메서드로 만들어져요.

struct 필드처럼 map 키나 값, list 요소는 중첩 필드로 추적돼요. 중첩 필드는 필드 ID와 null 허용 여부를 추적해요.

struct 필드는 NestedField.optional 또는 NestedField.required로 만들어요. map 값과 list 요소의 null 허용 여부는 map과 list 팩토리 메서드에서 설정돼요.

// struct<1 id: int, 2 data: optional string>
StructType struct = Struct.of(
    Types.NestedField.required(1, "id", Types.IntegerType.get()),
    Types.NestedField.optional(2, "data", Types.StringType.get())
  )

// map<1 key: int, 2 value: optional string>
MapType map = MapType.ofOptional(
    1, 2,
    Types.IntegerType.get(),
    Types.StringType.get()
  )

// array<1 element: int>
ListType list = ListType.ofRequired(1, IntegerType.get());

표현식 (Expressions)

아이스버그의 표현식은 테이블 스캔을 구성하는 데 사용돼요. 표현식을 만들려면 Expressions의 팩토리 메서드를 사용해요.

지원되는 조건(prerdicate) 표현식:

• isNull
• notNull
• equal
• notEqual
• lessThan
• lessThanOrEqual
• greaterThan
• greaterThanOrEqual
• in
• notIn
• startsWith
• notStartsWith

지원되는 표현식 연산:

• and
• or
• not

상수 표현식:

• alwaysTrue
• alwaysFalse

표현식 바인딩 (Expression binding)

표현식은 만들어지면 바인딩되지 않아요(unbound). 표현식을 사용하기 전에 데이터 타입에 바인딩해서 표현식 이름이 나타내는 필드 ID를 찾고, 조건 리터럴을 변환해요.

예를 들어 lessThan("x", 10) 표현식을 사용하기 전에, 아이스버그는 "x"가 어떤 컬럼을 가리키는지 결정하고 10을 그 컬럼의 데이터 타입으로 변환해야 해요.

표현식은 struct<1 x: long, 2 y: long> 타입 또는 struct<11 x: int, 12 y: int> 타입에 바인딩될 수 있어요.

표현식 예시 (Expression example)

table.newScan()
    .filter(Expressions.greaterThanOrEqual("x", 5))
    .filter(Expressions.lessThan("x", 10))

모듈 (Modules)

아이스버그 테이블 지원은 라이브러리 모듈로 구성돼 있어요.

• iceberg-common: 다른 모듈에서 사용하는 유틸리티 클래스 포함
• iceberg-api: 표현식, 타입, 테이블, 연산을 포함한 공개 아이스버그 API 포함
• iceberg-arrow: Apache Arrow를 인메모리 데이터 포맷으로 사용해 아이스버그 테이블에 저장된 데이터를 읽고 쓰는 아이스버그 타입 시스템 구현
• iceberg-aws: AWS S3에 저장되는 테이블 및/또는 AWS Glue 데이터 카탈로그로 정의되는 테이블에 사용할 아이스버그 API 구현 포함
• iceberg-core: 아이스버그 API 구현과 Avro 데이터 파일 지원 포함. 처리 엔진이 의존해야 하는 것
• iceberg-parquet: Parquet 파일로 뒷받침되는 테이블을 작업하기 위한 선택 모듈
• iceberg-orc: ORC 파일로 뒷받침되는 테이블을 작업하기 위한 선택 모듈 (실험적)
• iceberg-hive-metastore: Hive metastore Thrift 클라이언트로 뒷받침되는 아이스버그 테이블 구현

이 프로젝트 아이스버그는 처리 엔진과 관련 도구에 아이스버그 지원을 추가하는 모듈도 있어요.

• iceberg-spark: 아이스버그를 위한 스파크 Datasource V2 API 구현. 스파크 버전별 서브모듈 포함(shaded 버전에는 런타임 jar 사용)
• iceberg-flink: 아이스버그를 위한 Flink Table 및 DataStream API 구현(shaded 버전에는 iceberg-flink-runtime 사용)
• iceberg-mr: 아이스버그를 위한 MapReduce 및 Hive InputFormat·SerDe 구현(Hive에서 사용하려면 shaded 버전으로 iceberg-hive-runtime 사용)
• iceberg-nessie: 아이스버그 테이블 메타데이터 이력과 연산을 Project Nessie와 통합하는 데 사용되는 모듈
• iceberg-data: JVM 애플리케이션에서 아이스버그 테이블을 읽는 데 사용되는 클라이언트 라이브러리
• iceberg-runtime: 스파크가 아이스버그 테이블과 통합할 수 있도록 shaded 런타임 jar를 생성

더 알아보기 (Learn more)

• 자바 API 퀵스타트 (Java API quickstart)
• 스키마 (Schemas)
• 아이스버그 스펙 (Iceberg spec)