자바 API 퀵스타트
자바 API 퀵스타트 (Java API Quickstart)
이 문서에서는 아이스버그 자바 API를 사용해서 테이블을 만들고, 스키마와 파티션 스펙을 정의하고, 브랜치와 태그를 관리하는 방법을 알려드릴게요. Hive 카탈로그와 Hadoop 카탈로그를 사용해 테이블을 생성하고 로드하는 기본적인 흐름부터, 파티셔닝, 브랜칭·태깅의 실제 코드 예시까지 차례로 살펴볼게요.
출처: 문서
본문
테이블 만들기 (Create a table)
테이블은 Catalog 또는 Tables 인터페이스의 구현을 사용해서 만들어요.
Hive 카탈로그 사용하기 (Using a Hive catalog)
Hive 카탈로그는 Hive metastore에 연결해서 아이스버그 테이블을 추적해요. 이름과 몇 가지 속성으로 Hive 카탈로그를 초기화할 수 있어요. (참고: 카탈로그 속성)
import java.util.HashMap;
import java.util.Map;
import org.apache.iceberg.hive.HiveCatalog;
HiveCatalog catalog = new HiveCatalog();
catalog.setConf(spark.sparkContext().hadoopConfiguration()); // Optionally use Spark's Hadoop configuration
Map <String, String> properties = new HashMap<String, String>();
properties.put("warehouse", "...");
properties.put("uri", "...");
catalog.initialize("hive", properties);
HiveCatalog는 createTable, loadTable, renameTable, dropTable 같은 테이블 작업 메서드를 정의하는 Catalog 인터페이스를 구현해요. 테이블을 만들려면 Identifier와 Schema를 다른 초기 메타데이터와 함께 전달해요.
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
TableIdentifier name = TableIdentifier.of("logging", "logs");
Table table = catalog.createTable(name, schema, spec);
// or to load an existing table, use the following line
Table table = catalog.loadTable(name);
테이블의 스키마와 파티션 스펙은 아래에서 만들어져요.
Hadoop 카탈로그 사용하기 (Using a Hadoop catalog)
Hadoop 카탈로그는 Hive MetaStore에 연결할 필요가 없지만, 원자적 이름 변경(atomic rename)을 지원하는 HDFS 또는 유사한 파일 시스템에서만 사용할 수 있어요. Hadoop 카탈로그로 로컬 FS나 S3에서 동시 쓰기는 안전하지 않아요. Hadoop 카탈로그를 만들려면:
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.hadoop.HadoopCatalog;
Configuration conf = new Configuration();
String warehousePath = "hdfs://host:8020/warehouse_path";
HadoopCatalog catalog = new HadoopCatalog(conf, warehousePath);
Hive 카탈로그와 마찬가지로 HadoopCatalog도 Catalog를 구현하므로 createTable, loadTable, dropTable 같은 테이블 작업 메서드가 있어요.
이 예시는 Hadoop 카탈로그로 테이블을 만들어요.
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
TableIdentifier name = TableIdentifier.of("logging", "logs");
Table table = catalog.createTable(name, schema, spec);
// or to load an existing table, use the following line
Table table = catalog.loadTable(name);
테이블의 스키마와 파티션 스펙은 아래에서 만들어져요.
스파크의 테이블 (Tables in Spark)
스파크는 HiveCatalog를 사용해서 이름으로 테이블을 작업할 수 있어요.
// spark.sql.catalog.hive_prod = org.apache.iceberg.spark.SparkCatalog
// spark.sql.catalog.hive_prod.type = hive
spark.table("logging.logs");
스파크는 HadoopCatalog가 만든 테이블을 경로로도 로드할 수 있어요.
spark.read.format("iceberg").load("hdfs://host:8020/warehouse_path/logging/logs");
스키마 (Schemas)
스키마 만들기 (Create a schema)
이 예시는 logs 테이블의 스키마를 만들어요.
import org.apache.iceberg.Schema;
import org.apache.iceberg.types.Types;
Schema schema = new Schema(
Types.NestedField.required(1, "level", Types.StringType.get()),
Types.NestedField.required(2, "event_time", Types.TimestampType.withZone()),
Types.NestedField.required(3, "message", Types.StringType.get()),
Types.NestedField.optional(4, "call_stack", Types.ListType.ofRequired(5, Types.StringType.get()))
);
아이스버그 API를 직접 사용할 때는 타입 ID가 필요해요. 스파크, Avro, Parquet 같은 다른 스키마 포맷에서의 변환은 새 ID를 자동으로 할당해요.
테이블이 만들어질 때 스키마의 모든 ID는 고유성을 보장하기 위해 다시 할당돼요.
Avro에서 스키마 변환 (Convert a schema from Avro)
기존 Avro 스키마에서 아이스버그 스키마를 만들려면 AvroSchemaUtil의 변환기를 사용해요.
import org.apache.avro.Schema;
import org.apache.avro.Schema.Parser;
import org.apache.iceberg.avro.AvroSchemaUtil;
Schema avroSchema = new Parser().parse("{\"type\": \"record\" , ... }");
Schema icebergSchema = AvroSchemaUtil.toIceberg(avroSchema);
스파크에서 스키마 변환 (Convert a schema from Spark)
기존 테이블에서 아이스버그 스키마를 만들려면 SparkSchemaUtil의 변환기를 사용해요.
import org.apache.iceberg.spark.SparkSchemaUtil;
Schema schema = SparkSchemaUtil.schemaForTable(sparkSession, tableName);
파티셔닝 (Partitioning)
파티션 스펙 만들기 (Create a partition spec)
파티션 스펙은 아이스버그가 레코드를 데이터 파일로 그룹화하는 방법을 설명해요. 파티션 스펙은 빌더를 사용해서 테이블의 스키마에 대해 만들어져요.
이 예시는 로그 이벤트 타임스탬프의 시(hour)와 로그 레벨로 레코드를 파티셔닝하는 logs 테이블의 파티션 스펙을 만들어요.
import org.apache.iceberg.PartitionSpec;
PartitionSpec spec = PartitionSpec.builderFor(schema)
.hour("event_time")
.identity("level")
.build();
아이스버그가 제공하는 다양한 파티션 변환에 대한 자세한 내용은 이 페이지를 참고해주세요.
브랜칭과 태깅 (Branching and Tagging)
브랜치와 태그 만들기 (Creating branches and tags)
새 브랜치와 태그는 자바 라이브러리의 ManageSnapshots API로 만들 수 있어요.
/* Create a branch test-branch which is retained for 1 week, and the latest 2 snapshots on test-branch will always be retained.
Snapshots on test-branch which are created within the last hour will also be retained. */
String branch = "test-branch";
table.manageSnapshots()
.createBranch(branch, 3)
.setMinSnapshotsToKeep(branch, 2)
.setMaxSnapshotAgeMs(branch, 3600000)
.setMaxRefAgeMs(branch, 604800000)
.commit();
// Create a tag historical-tag at snapshot 10 which is retained for a day
String tag = "historical-tag"
table.manageSnapshots()
.createTag(tag, 10)
.setMaxRefAgeMs(tag, 86400000)
.commit();
브랜치에 커밋하기 (Committing to branches)
브랜치에 쓰는 것은 연산에서 toBranch를 지정해서 수행할 수 있어요. 전체 목록은 UpdateOperations를 참고해주세요.
// Append FILE_A to branch test-branch
String branch = "test-branch";
table.newAppend()
.appendFile(FILE_A)
.toBranch(branch)
.commit();
// Perform row level updates on "test-branch"
table.newRowDelta()
.addRows(DATA_FILE)
.addDeletes(DELETES)
.toBranch(branch)
.commit();
// Perform a rewrite operation replacing SMALL_FILE_1 and SMALL_FILE_2 on "test-branch" with compactedFile.
table.newRewrite()
.rewriteFiles(ImmutableSet.of(SMALL_FILE_1, SMALL_FILE_2), ImmutableSet.of(compactedFile))
.toBranch(branch)
.commit();
브랜치와 태그에서 읽기 (Reading from branches and tags)
브랜치나 태그에서 읽는 것은 Table Scan API에서 useRef API에 브랜치나 태그를 전달하는 방식으로 평소처럼 할 수 있어요. 브랜치를 전달하면 사용되는 스냅샷은 그 브랜치의 헤드예요. 현재는 브랜치에서 읽으면서 스캔에 asOfSnapshotId를 지정하는 것은 지원되지 않는다는 점에 유의해주세요.
// Read from the head snapshot of test-branch
TableScan branchRead = table.newScan().useRef("test-branch");
// Read from the snapshot referenced by audit-tag
TableScan tagRead = table.newScan().useRef("audit-tag");
브랜치와 태그 교체 및 패스트포워드 (Replacing and fast forwarding branches and tags)
기존 브랜치와 태그가 가리키는 스냅샷은 replace API로 업데이트할 수 있어요. fast forward 연산은 git fast-forwarding과 비슷해요. 대상 브랜치가 소스 브랜치의 조상일 때 소스 브랜치나 태그의 헤드로 대상 브랜치를 전진시키는 데 fast forward를 사용할 수 있어요. fast forward와 replace 모두에서, 대상 브랜치의 보존 속성은 기본적으로 유지돼요.
// Update "test-branch" to point to snapshot 4
table.manageSnapshots()
.replaceBranch(branch, 4)
.commit()
String tag = "audit-tag";
// Replace "audit-tag" to point to snapshot 3 and update its retention
table.manageSnapshots()
.replaceBranch(tag, 4)
.setMaxRefAgeMs(1000)
.commit()
보존 속성 업데이트 (Updating retention properties)
브랜치와 태그의 보존 속성도 업데이트할 수 있어요. 브랜치나 태그 자체의 보존 속성을 업데이트하려면 setMaxRefAgeMs를 사용해요. 브랜치 스냅샷 보존 속성은 setMinSnapshotsToKeep과 setMaxSnapshotAgeMs API로 업데이트할 수 있어요.
String branch = "test-branch";
// Update retention properties for test-branch
table.manageSnapshots()
.setMinSnapshotsToKeep(branch, 10)
.setMaxSnapshotAgeMs(branch, 7200000)
.setMaxRefAgeMs(branch, 604800000)
.commit();
// Update retention properties for test-tag
table.manageSnapshots()
.setMaxRefAgeMs("test-tag", 604800000)
.commit();
브랜치와 태그 제거 (Removing branches and tags)
브랜치와 태그는 각각 removeBranch와 removeTag API로 제거할 수 있어요.
// Remove test-branch
table.manageSnapshots()
.removeBranch("test-branch")
.commit()
// Remove test-tag
table.manageSnapshots()
.removeTag("test-tag")
.commit()