Iceberg 커넥터

Iceberg 커넥터 (Iceberg connector)

Apache Iceberg는 대규모 분석 데이터셋을 위한 오픈 테이블 포맷이에요. Iceberg 커넥터는 Iceberg Table Spec에 정의된 Iceberg 포맷으로 작성된 파일에 저장된 데이터를 질의할 수 있게 해 줘요. 이 커넥터는 Apache Iceberg 테이블 스펙 버전 1과 2를 지원해요. 포맷 버전 3 지원은 실험적이에요.

출처: 문서

본문

테이블 상태는 메타데이터 파일에 유지돼요. 테이블 상태의 모든 변경은 새 메타데이터 파일을 만들고 원자적 교체(atomic swap)로 기존 메타데이터를 대체해요. 테이블 메타데이터 파일은 테이블 스키마, 파티셔닝 구성, 커스텀 속성, 테이블 콘텐츠의 스냅샷을 추적해요.

Iceberg 데이터 파일은 테이블 정의의 format 속성에 따라 Parquet, ORC, Avro 포맷 중 하나로 저장돼요.

Iceberg는 Hive의 알려진 확장성 한계를 개선하도록 설계됐어요. Hive는 MySQL 같은 관계형 데이터베이스로 지원되는 메타스토어에 테이블 메타데이터를 저장하죠. Hive는 파티션 위치는 메타스토어에서 추적하지만 개별 데이터 파일은 추적하지 않아요. Hive 커넥터를 사용하는 Trino 쿼리는 먼저 메타스토어를 호출해 파티션 위치를 얻고, 그다음 기본 파일 시스템을 호출해 각 파티션 안의 모든 데이터 파일을 나열하고, 각 데이터 파일의 메타데이터를 읽어야 해요.

Iceberg는 데이터 파일의 경로를 메타데이터 파일에 저장하므로, 읽어야 하는 파일에 대해서만 기본 파일 시스템을 참조해요.

요구사항 (Requirements)

Iceberg를 사용하려면 다음이 필요해요:

• Trino 코디네이터와 워커에서 분산 오브젝트 스토리지로 네트워크 접근.
• Hive 메타스토어 서비스(HMS), AWS Glue 카탈로그, JDBC 카탈로그, REST 카탈로그, Nessie 서버, 또는 Snowflake 카탈로그 접근.
• 지원되는 파일 시스템에 Parquet(기본), ORC, 또는 Avro 파일 포맷으로 저장된 데이터 파일.

일반 설정 (General configuration)

Iceberg 커넥터를 구성하려면 iceberg 커넥터를 참조하는 etc/catalog/example.properties 카탈로그 속성 파일을 만드세요.

Hive 메타스토어 카탈로그가 기본 구현이에요.

지원되는 파일 시스템 중 하나를 선택해 구성해야 해요.

connector.name=iceberg
hive.metastore.uri=thrift://example.net:9083
fs.x.enabled=true

스키마·테이블 생성

위치 지정 스키마(S3, HDFS 등):

CREATE SCHEMA example.example_s3_schema
WITH (location = 's3://my-bucket/a/path/');

CREATE SCHEMA example.example_s3a_schema
WITH (location = 's3a://my-bucket/a/path/');

CREATE SCHEMA example.example_hdfs_schema
WITH (location='hdfs://hadoop-master:9000/user/hive/warehouse/a/path/');

CREATE SCHEMA example.example_hdfs_schema;

파티셔닝·정렬·위치를 지정한 테이블:

CREATE TABLE example_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE
)
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['c1', 'c2'],
    sorted_by = ARRAY['c3'],
    location = 's3://my-bucket/a/path/'
);

데이터를 채운 테이블 생성:

CREATE TABLE tiny_nation
WITH (
    format = 'PARQUET'
)
AS
    SELECT *
    FROM nation
    WHERE nationkey < 10;

값 리스트로 채운 테이블:

CREATE TABLE yearly_clicks (
    year,
    clicks
)
WITH (
    partitioning = ARRAY['year']
)
AS VALUES
    (2021, 10000),
    (2022, 20000);

시스템 프로시저 (System procedures)

메타스토어에 없는 기존 Iceberg 테이블 등록:

CALL example.system.register_table(
  schema_name => 'testdb',
  table_name => 'customer_orders',
  table_location => 'hdfs://hadoop-master:9000/user/hive/warehouse/customer_orders-581fad8517934af6be1857a903559d44');

특정 메타데이터 파일을 지정해 등록:

CALL example.system.register_table(
  schema_name => 'testdb',
  table_name => 'customer_orders',
  table_location => 'hdfs://hadoop-master:9000/user/hive/warehouse/customer_orders-581fad8517934af6be1857a903559d44',
  metadata_file_name => '00003-409702ba-4735-4645-8f14-09537cc0b2c8.metadata.json');

등록 해제:

CALL example.system.unregister_table(
  schema_name => 'testdb',
  table_name => 'customer_orders');

기존 Hive 테이블을 Iceberg로 마이그레이션:

CALL example.system.migrate(
    schema_name => 'testdb',
    table_name => 'customer_orders');

재귀 디렉토리 검색으로 마이그레이션:

CALL example.system.migrate(
    schema_name => 'testdb',
    table_name => 'customer_orders',
    recursive_directory => 'true');

파일 추가 (Add files)

테이블이 아닌 디렉토리의 데이터를 Iceberg 테이블에 추가할 수 있어요. 다른 테이블의 파일 추가:

ALTER TABLE example.lakehouse.iceberg_customer_orders
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders');

카탈로그 컨텍스트 내에서:

USE example.lakehouse;
ALTER TABLE iceberg_customer_orders
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders');

파티션 필터 지정:

ALTER TABLE example.lakehouse.iceberg_customer_orders
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders',
    partition_filter => map(ARRAY['region', 'country'], ARRAY['ASIA', 'JAPAN']));

재귀 검색:

ALTER TABLE example.lakehouse.iceberg_customer_orders
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders',
    recursive_directory => 'true');

임의 위치의 파일 추가:

ALTER TABLE example.lakehouse.iceberg_customer_orders
EXECUTE add_files(
    location => 's3://my-bucket/a/path',
    format => 'ORC');

함수 (Functions)

Iceberg 커넥터는 system.bucket 함수를 제공해요:

SELECT system.bucket('trino', 16);

버킷 파티셔닝 조건:

SELECT count(*)
FROM customer
WHERE system.bucket(custkey, 16) = 2;

삭제 (Delete)

DELETE FROM example.testdb.customer_orders
WHERE country = 'US';

파일·메타데이터 유지보수 (Maintenance)

파일 병합 최적화:

ALTER TABLE test_table EXECUTE optimize

        metric_name         | metric_value
----------------------------+--------------
 rewritten_data_files_count |            1
 removed_delete_files_count |            1
 added_data_files_count     |            2

크기 임계값:

ALTER TABLE test_table EXECUTE optimize(file_size_threshold => '128MB')

파티션 조건:

ALTER TABLE test_partitioned_table EXECUTE optimize
WHERE partition_key = 1

시간 조건:

ALTER TABLE test_table EXECUTE optimize
WHERE CAST(timestamp_tz AS DATE) > DATE '2021-12-31'

파일 수정 시간 기준:

ALTER TABLE test_table EXECUTE optimize
WHERE "$file_modified_time" > date_trunc('day', CURRENT_TIMESTAMP);

매니페스트 병합:

ALTER TABLE test_table EXECUTE optimize_manifests;

metric_name                      | metric_value
---------------------------------+--------------
rewritten_manifests_count        |            2
added_manifests_count            |            1
kept_manifests_count             |            1
processed_manifest_entries_count |            2

오래된 스냅샷 만료:

ALTER TABLE test_table EXECUTE expire_snapshots(retention_threshold => '7d');

고아 파일 제거:

ALTER TABLE test_table EXECUTE remove_orphan_files(retention_threshold => '7d');

        metric_name         | metric_value
----------------------------+--------------
 processed_manifests_count  |            2
 active_files_count         |           98
 scanned_files_count        |           97
 deleted_files_count        |            0
 deleted_bytes              |            0

확장 통계 삭제:

ALTER TABLE test_table EXECUTE drop_extended_stats;

테이블 속성 변경

포맷 버전 변경:

ALTER TABLE table_name SET PROPERTIES format_version = 2;

파티셔닝 변경:

ALTER TABLE table_name SET PROPERTIES partitioning = ARRAY[, 'my_new_partition_column'];

속성이 있는 테이블 생성(Parquet):

CREATE TABLE test_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE)
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['c1', 'c2'],
    location = '/var/example_tables/test_table');

블룸 필터가 있는 ORC:

CREATE TABLE test_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE)
WITH (
    format = 'ORC',
    compression_codec = 'SNAPPY',
    location = '/var/example_tables/test_table',
    orc_bloom_filter_columns = ARRAY['c1', 'c2'],
    orc_bloom_filter_fpp = 0.05);

중첩 컬럼 파티셔닝(Avro):

CREATE TABLE test_table (
    data INTEGER,
    parent ROW(child1 DOUBLE, child2 INTEGER))
WITH (
    format = 'AVRO',
    partitioning = ARRAY['"parent.child1"']);

메타데이터 테이블

Iceberg는 $ 접미사 메타데이터 테이블을 제공해요. $properties:

SELECT * FROM "test_table$properties";

 key                   | value    |
-----------------------+----------+
write.format.default   | PARQUET  |

$history — 스냅샷 기록:

SELECT * FROM "test_table$history";

 made_current_at                  | snapshot_id          | parent_id            | is_current_ancestor
----------------------------------+----------------------+----------------------+--------------------
2022-01-10 08:11:20 Europe/Vienna | 8667764846443717831  |                |  true
2022-01-10 08:11:34 Europe/Vienna | 7860805980949777961  | 8667764846443717831  |  true

$metadata_log_entries — 메타데이터 로그:

SELECT * FROM "test_table$metadata_log_entries";

$snapshots — 스냅샷 목록:

SELECT * FROM "test_table$snapshots";

$manifests — 매니페스트 목록:

SELECT * FROM "test_table$manifests";

$partitions — 파티션 요약:

SELECT * FROM "test_table$partitions";

 partition             | record_count  | file_count    | total_size    |  data
-----------------------+---------------+---------------+---------------+------------------------------------------------------
{c1=1, c2=2021-01-12}  |  2            | 2             |  884          | {c3={min=1.0, max=2.0, null_count=0, nan_count=NULL}}
{c1=1, c2=2021-01-13}  |  1            | 1             |  442          | {c3={min=1.0, max=1.0, null_count=0, nan_count=NULL}}

$files — 데이터 파일 상세:

SELECT * FROM "test_table$files";

$entries — 모든 매니페스트 항목:

SELECT * FROM "test_table$entries";

$refs — 태그·브랜치 참조:

SELECT * FROM "test_table$refs";

name            | type   | snapshot_id | max_reference_age_in_ms | min_snapshots_to_keep | max_snapshot_age_in_ms |
----------------+--------+-------------+-------------------------+-----------------------+------------------------+
example_tag     | TAG    | 10000000000 | 10000                   | null                  | null                   |
example_branch  | BRANCH | 20000000000 | 20000                   | 2                     | 30000                  |

test_materialized_view$files 같은 메타데이터 테이블:

SELECT * FROM "test_materialized_view$files";

하이든 컬럼 (Hidden columns)

$partition, $path, $file_modified_time 하이든 컬럼:

SELECT *, "$partition", "$path", "$file_modified_time"
FROM example.web.page_views;

SELECT *
FROM example.web.page_views
WHERE "$path" = '/usr/iceberg/table/web.page_views/data/file_01.parquet'

SELECT *
FROM example.web.page_views
WHERE "$file_modified_time" = CAST('2022-07-01 01:02:03.456 UTC' AS TIMESTAMP WITH TIME ZONE)

헬퍼 테이블 (Helper table)

system.iceberg_tables로 Iceberg 테이블을 나열할 수 있어요:

SELECT * FROM example.system.iceberg_tables;

 table_schema | table_name  |
--------------+-------------+
 tpcds        | store_sales |
 tpch         | nation      |
 tpch         | region      |
 tpch         | orders      |

파티셔닝 변환 (Partition transforms)

Iceberg 파티셔닝 변환을 테이블 생성에 쓸 수 있어요. month, bucket, 컬럼 직접 지정:

CREATE TABLE example.testdb.customer_orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (partitioning = ARRAY['month(order_date)', 'bucket(account_number, 10)', 'country']);

정렬 (Sorting)

sorted_by로 정렬 지정:

CREATE TABLE example.customers.orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (sorted_by = ARRAY['order_date']);

정렬 방향·NULL 순서 지정:

CREATE TABLE example.customers.orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (sorted_by = ARRAY['order_date DESC NULLS FIRST', 'order_id ASC NULLS LAST']);

파티셔닝과 정렬 동시 지정:

CREATE TABLE example.customers.orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (
    partitioning = ARRAY['month(order_date)'],
    sorted_by = ARRAY['order_date']
);

시간 여행 (Time travel)

최신 스냅샷 id 확인:

SELECT snapshot_id
FROM example.testdb."customer_orders$snapshots"
ORDER BY committed_at DESC;

스냅샷 버전으로 조회:

SELECT *
FROM example.testdb.customer_orders FOR VERSION AS OF 8954597067493422955;

타임스탬프로 조회:

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 09:59:29.803 Europe/Vienna';

과거 스냅샷으로 테이블 재생성:

CREATE OR REPLACE TABLE example.testdb.customer_orders AS
SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 09:59:29.803 Europe/Vienna';

날짜·정밀도 지정:

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF DATE '2022-03-23';

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 00:00:00';

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 00:00:00.000 Europe/Vienna';

태그·브랜치 이름으로 조회:

SELECT *
FROM example.testdb.customer_orders FOR VERSION AS OF 'historical-tag';

SELECT *
FROM example.testdb.customer_orders FOR VERSION AS OF 'test-branch';

롤백 (Rollback)

최신 스냅샷 id를 구하고:

SELECT snapshot_id
FROM example.testdb."customer_orders$snapshots"
ORDER BY committed_at DESC LIMIT 1;

스냅샷으로 롤백:

ALTER TABLE testdb.customer_orders EXECUTE rollback_to_snapshot(8954597067493422955);

체인지 데이터 피드 (Change Data Feed)

table_changes 테이블 함수:

SELECT
  *
FROM
  TABLE(
    system.table_changes(
      schema_name => 'default',
      table_name => 't1',
      start_snapshot_id => 6541165659943306573,
      end_snapshot_id => 6745790645714043599
    )
  );

예시 — 페이지 테이블을 만들고:

CREATE TABLE test_schema.pages (page_url VARCHAR, domain VARCHAR, views INTEGER);

데이터 삽입:

INSERT INTO test_schema.pages
    VALUES
        ('url1', 'domain1', 1),
        ('url2', 'domain2', 2),
        ('url3', 'domain1', 3);
INSERT INTO test_schema.pages
    VALUES
        ('url4', 'domain1', 400),
        ('url5', 'domain2', 500),
        ('url6', 'domain3', 2);

스냅샷을 확인하고:

SELECT
    snapshot_id,
    parent_id,
    operation
FROM test_schema."pages$snapshots";

     snapshot_id     |      parent_id      | operation
---------------------+---------------------+-----------
 2009020668682716382 |                NULL | append
 2135434251890923160 | 2009020668682716382 | append
 3108755571950643966 | 2135434251890923160 | append
(3 rows)

변경 피드를 조회:

SELECT
    *
FROM
    TABLE(
            system.table_changes(
                    schema_name => 'test_schema',
                    table_name => 'pages',
                    start_snapshot_id => 2009020668682716382,
                    end_snapshot_id => 3108755571950643966
            )
    )
ORDER BY _change_ordinal ASC;

 page_url | domain  | views | _change_type | _change_version_id  |      _change_timestamp      | _change_ordinal
----------+---------+-------+--------------+---------------------+-----------------------------+-----------------
 url1     | domain1 |     1 | insert       | 2135434251890923160 | 2024-04-04 21:24:26.105 UTC |               0
 url2     | domain2 |     2 | insert       | 2135434251890923160 | 2024-04-04 21:24:26.105 UTC |               0
 url3     | domain1 |     3 | insert       | 2135434251890923160 | 2024-04-04 21:24:26.105 UTC |               0
 url4     | domain1 |   400 | insert       | 3108755571950643966 | 2024-04-04 21:24:28.318 UTC |               1
 url5     | domain2 |   500 | insert       | 3108755571950643966 | 2024-04-04 21:24:28.318 UTC |               1
 url6     | domain3 |     2 | insert       | 3108755571950643966 | 2024-04-04 21:24:28.318 UTC |               1
(6 rows)

통계 (Statistics)

ANALYZE table_name;

컬럼 지정:

ANALYZE table_name WITH (columns = ARRAY['col_1', 'col_2']);

멀티 카탈로그 액세스

기본 카탈로그와 다른 카탈로그의 테이블을 참조하려면 정규화된 catalog.schema.table 이름을 쓰세요:

USE example.example_schema;

EXPLAIN SELECT * FROM example_table;

                               Query Plan
-------------------------------------------------------------------------
Fragment 0 [SOURCE]
     ...
     Output[columnNames = [...]]
     │   ...
     └─ TableScan[table = another_catalog:example_schema:example_table]
            ...

EXPLAIN SELECT * FROM example.example_schema.example_table;

더 알아보기 (Learn more)

다른 데이터 레이크 포맷 커넥터가 궁금하다면 Delta Lake 커넥터 문서를 이어서 읽어 보세요.