[DATA] Apache iceberg

 

Hive and Iceberg Quickstart - Apache Iceberg™

 

 

 

 

Iceberg?

• 데이터 버전 관리: Apache Iceberg는 사용자가 시간 경과에 따른 데이터의 변경 내용을 추적할 수 있는 데이터 버전 관리를 지원합니다. 따라서 사용자가 이전 버전의 데이터에 액세스하여 쿼리하고, 업데이트 작업과 삭제 작업 간에 변경된 데이터를 분석하는 시간 이동 기능이 지원됩니다.

• 증분 처리: Iceberg는 사용자가 마지막 실행 이후 변경된 데이터만 처리할 수 있도록 하는 증분 처리 기능을 지원합니다. 이 기능을 변경 데이터 캡처(CDC)라고도 합니다. 이 기능은 데이터 처리 효율성과 성능을 개선하는 데 도움이 될 수 있습니다.

 

ETL, MART의 Data가 어떤 과정을 거쳐 인입되었는 지 확인이 된다면, 부담이 덜 할 듯 싶다.

 

create 

CREATE TABLE nyc.taxis
(
  trip_id bigint,
  trip_distance float,
  fare_amount double,
  store_and_fwd_flag string
)
PARTITIONED BY (vendor_id bigint) STORED BY ICEBERG;

-- as SELECT 
CREATE TABLE target PARTITIONED BY SPEC (year(year_field), identity_field) STORED BY ICEBERG AS
    SELECT * FROM source;
    
    
-- use hive catalog 
SET
iceberg.catalog.hadoop_cat.type=hadoop;
SET
iceberg.catalog.hadoop_cat.warehouse=hdfs://example.com:8020/hadoop_cat;

CREATE
EXTERNAL TABLE database_a.table_a
STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
TBLPROPERTIES ('iceberg.catalog'='hadoop_cat');

 

 

 

branch를 통한 테이블 관리 

-- CREATE branch1 with default properties.
ALTER TABLE test CREATE BRANCH branch1;

-- CREATE branch1 at a specific snapshot ID.
ALTER TABLE test CREATE BRANCH branch1 FOR SYSTEM_VERSION AS OF 3369973735913135680;

-- CREATE branch1 using system time.
ALTER TABLE test CREATE BRANCH branch1 FOR SYSTEM_TIME AS OF '2023-09-16 09:46:38.939 Etc/UTC';

-- CREATE branch1 with a specified number of snapshot retentions.
ALTER TABLE test CREATE BRANCH branch1 FOR SYSTEM_VERSION AS OF 3369973735913135680 WITH SNAPSHOT RETENTION 5 SNAPSHOTS;

-- CREATE branch1 using a specific tag.
ALTER TABLE test CREATE BRANCH branch1 FOR TAG AS OF tag1;

-- DROP branch1
ALTER TABLE test DROP BRANCH branch1;

-- DROP branch1 IF EXISTS
ALTER TABLE test DROP BRANCH IF EXISTS branch1;

# 특정 브랜치를 지정해서 조회 가능
-- Branches should be specified as <database_name>.<table_name>.branch_<branch_name>
SELECT * FROM default.test.branch_branch1;

# 특정 브랜치를 지정해서 DML 가능
-- Branches should be specified as <database_name>.<table_name>.branch_<branch_name>
INSERT INTO default.test.branch_branch1
VALUES ('a', 1);
INSERT INTO default.test.branch_branch1
SELECT...;

 

 

 

TIMETRAVEL

시뮬레이션이나, 수정 사항 작업 시 유용. 

 

Hive supports snapshot id based and time base timetravel queries. For these views it is possible to use projections / joins / filters / etc. The function is available with the following syntax:

SELECT * FROM table_a FOR SYSTEM_TIME AS OF '2021-08-09 10:35:57';
SELECT * FROM table_a FOR SYSTEM_VERSION AS OF 1234567;

 

You can expire snapshots of an Iceberg table using an ALTER TABLE query from Hive. You should periodically expire snapshots to delete data files that is no longer needed, and reduce the size of table metadata.

Each write to an Iceberg table from Hive creates a new snapshot, or version, of a table. Snapshots can be used for time-travel queries, or the table can be rolled back to any valid snapshot. Snapshots accumulate until they are expired by the expire_snapshots operation. Enter a query to expire snapshots having the following timestamp: 2021-12-09 05:39:18.689000000

ALTER TABLE test_table EXECUTE expire_snapshots('2021-12-09 05:39:18.689000000')

 

스냅샷의 만료 시간을 기록 할 수 있다. ( 데이터 용량 관리에 유용)

 

 

 

TRINO

-- You must select and configure one of the supported file systems.

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

 

 

TABLE 관리 : 

CREATE TABLE example.web.page_views (
  view_time TIMESTAMP,
  user_id BIGINT,
  page_url VARCHAR,
  ds DATE,
  country VARCHAR
)
WITH (
  format = 'ORC',
  partitioned_by = ARRAY['ds', 'country'],
  bucketed_by = ARRAY['user_id'],
  bucket_count = 50
)

또는 S3 bucket

CREATE SCHEMA example.web
WITH (location = 's3://my-bucket/')

SCHEMA - TABLE

CREATE TABLE example.web.request_logs (
  request_time TIMESTAMP,
  url VARCHAR,
  ip VARCHAR,
  user_agent VARCHAR
)
WITH (
  format = 'TEXTFILE',
  external_location = 's3://my-bucket/data/logs/'
)