使用 Trino 處理 Iceberg 資料表 - AWS 方案指引
EC2 上的 Amazon EMR 設定建立 Iceberg 資料表從 Iceberg 資料表讀取將資料升級至 Iceberg 資料表從 Iceberg 資料表刪除記錄查詢 Iceberg 資料表中繼資料使用時間歷程搭配 Trino 使用 Iceberg 時的考量事項

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

使用 Trino 處理 Iceberg 資料表

本節說明如何在 Amazon EMR 上使用 Trino 來設定和操作 Iceberg 資料表。這些範例是樣板程式碼，您可以在 Amazon EMR on EC2 叢集上執行。本節中的程式碼範例和組態假設您使用的是 Amazon EMR 發行版本 emr-7.9.0。

EC2 上的 Amazon EMR 設定

  1. 使用下列內容建立 iceberg.properties檔案。如果未在 CREATE TABLE陳述式中明確指定格式，則 iceberg.file-format=parquet設定會決定新資料表的預設儲存格式。

    connector.name=iceberg
iceberg.catalog.type=glue
iceberg.file-format=parquet
fs.native-s3.enabled=true

  2. 將 iceberg.properties 檔案上傳至 S3 儲存貯體。 

  3. 建立從 S3 儲存貯體複製iceberg.properties檔案的引導操作，並將其儲存為您要建立的 Amazon EMR 叢集上的 Trino 組態檔案。請務必<S3-bucket-name>將 取代為您的 S3 儲存貯體名稱。 

    #!/bin/bash
set -ex
sudo aws s3 cp s3://<S3-bucket-name>/iceberg.properties /etc/trino/conf/catalog/iceberg.properties

  4. 建立已安裝 Trino 的 Amazon EMR 叢集，並將先前指令碼的執行指定為引導操作。以下是用於建立叢集的 sample AWS Command Line Interface (AWS CLI) 命令：

    aws emr create-cluster --release-label emr-7.9.0 \
--applications Name=Trino \
--region <region> \
--name Trino_Iceberg_Cluster \
--bootstrap-actions '[{"Path":"s3://<S3-bucket-name>/bootstrap.sh","Name":"Add iceberg.properties"}]' \
--instance-groups '[{"InstanceGroupType":"MASTER","InstanceCount":1,"InstanceType":"m5.xlarge"},{"InstanceGroupType":"CORE","InstanceCount":3,"InstanceType":"m5.xlarge"}]' \
--service-role "<IAM-service-role>" \
--ec2-attributes '{"KeyName":"<key-name>","InstanceProfile":"<EMR-EC2-instance-profile>"}'

    您取代的位置：

    • <S3-bucket-name> 使用您的 S3 儲存貯體名稱

    • <region> 您的特定 AWS 區域

    • <key-name> 您的金鑰對。如果金鑰對不存在，則會建立金鑰對。

    • <IAM-service-role> 搭配遵循最低權限原則的 Amazon EMR 服務角色。 

    • <EMR-EC2-instance-profile> 您的執行個體描述檔。 

  5. 初始化 Amazon EMR 叢集後，您可以執行下列命令來初始化 Trino 工作階段：

    trino-cli

  6. 在 Trino CLI 中，您可以執行下列動作來檢視目錄：

    SHOW CATALOGS;

建立 Iceberg 資料表

若要建立 Iceberg 資料表，您可以使用 CREATE TABLE陳述式。  以下是建立使用 Iceberg 隱藏分割的分割資料表的範例：

CREATE TABLE iceberg.iceberg_db.iceberg_table (
             userid int,
             firstname varchar,
             city varchar)
        WITH (
             format = 'PARQUET',
             partitioning = ARRAY['city', 'bucket(userid, 16)'],
             location = 's3://<S3-bucket>/<prefix>');
注意

如果您未指定格式，則會使用您在上一節中設定iceberg.file-format的值。

若要插入資料，請使用 INSERT INTO命令。範例如下：

INSERT INTO iceberg.iceberg_db.iceberg_table (userid, firstname, city)
VALUES 
    (1001, 'John', 'New York'),
    (1002, 'Mary', 'Los Angeles'),
    (1003, 'Mateo', 'Chicago'),
    (1004, 'Shirley', 'Houston'),
    (1005, 'Diego', 'Miami'),
    (1006, 'Nikki', 'Seattle'),
    (1007, 'Pat', 'Boston'),
    (1008, 'Terry', 'San Francisco'),
    (1009, 'Richard', 'Denver'),
    (1010, 'Pat', 'Phoenix');

從 Iceberg 資料表讀取

您可以使用 SELECT陳述式讀取 Iceberg 資料表的最新狀態，如下所示：

SELECT * FROM iceberg.iceberg_db.iceberg_table;

將資料升級至 Iceberg 資料表

您可以使用 MERGE INTO陳述式執行 upsert 操作 （同時插入新記錄並更新現有記錄）。範例如下：

MERGE INTO iceberg.iceberg_db.iceberg_table target
USING (
      VALUES 
        (1001, 'John Updated', 'Boston'),       -- Update existing user
        (1002, 'Mary Updated', 'Seattle'),      -- Update existing user
        (1011, 'Martha', 'Portland'),           -- Insert new user
        (1012, 'Paulo', 'Austin')               -- Insert new user
    ) AS source (userid, firstname, city)
   ON target.userid = source.userid
 WHEN MATCHED THEN
      UPDATE SET 
        firstname = source.firstname,
        city = source.city
 WHEN NOT MATCHED THEN
      INSERT (userid, firstname, city)
      VALUES (source.userid, source.firstname, source.city);

從 Iceberg 資料表刪除記錄

若要從 Iceberg 資料表刪除資料，請使用 DELETE FROM運算式，並指定符合要刪除資料列的篩選條件。範例如下：

DELETE FROM iceberg.iceberg_db.iceberg_table WHERE userid IN (1003, 1004);

查詢 Iceberg 資料表中繼資料

Iceberg 提供透過 SQL 存取其中繼資料的權限。您可以透過查詢命名空間 來存取任何指定資料表 (<table_name>) 的中繼資料"<table_name>.$<metadata_table>"。如需中繼資料資料表的完整清單，請參閱 Iceberg 文件中的檢查資料表。

以下是檢查 Iceberg 中繼資料的查詢清單範例：

SELECT  FROM iceberg.iceberg_db."iceberg_table$snapshots";
SELECT  FROM iceberg.iceberg_db."iceberg_table$history";
SELECT  FROM iceberg.iceberg_db."iceberg_table$partitions";
SELECT  FROM iceberg.iceberg_db."iceberg_table$files";
SELECT  FROM iceberg.iceberg_db."iceberg_table$manifests";
SELECT  FROM iceberg.iceberg_db."iceberg_table$refs";
SELECT * FROM iceberg.iceberg_db."iceberg_table$metadata_log_entries";

例如，這個查詢：

 SELECT * FROM iceberg.iceberg_db."iceberg_table$snapshots";

提供輸出：

查詢 Iceberg 資料表中繼資料的輸出。

使用時間歷程

Iceberg 資料表中的每個寫入操作 （插入、更新、 upsert 或刪除） 都會建立新的快照。然後，您可以將這些快照用於時間歷程，以返回時間並檢查過去資料表的狀態。

下列時間歷程查詢會根據特定 顯示資料表的狀態snapshot_id：

SELECT *
FROM iceberg.iceberg_db.iceberg_table FOR VERSION AS OF 241938428756831817;

下列時間歷程查詢會根據特定時間戳記顯示資料表的狀態：

SELECT *
FROM iceberg.iceberg_db.iceberg_table FOR TIMESTAMP AS OF TIMESTAMP '2025-05-28 16:09:40.268 UTC'

搭配 Trino 使用 Iceberg 時的考量事項

Iceberg 資料表上的 Trino 寫入操作遵循 merge-on-readdesign，因此它們會建立位置刪除檔案，而不是重寫受更新或刪除影響的整個資料檔案。如果您想要使用copy-on-write方法，請考慮使用 Spark 進行寫入操作。