EC2 での Amazon EMR のセットアップ Iceberg テーブルの作成 Iceberg テーブルからの読み取り Iceberg テーブルへのデータの更新 Iceberg テーブルからのレコードの削除 Iceberg テーブルデータのクエリの実行タイムトラベルの使用 Trino で Iceberg を使用する場合の考慮事項

Trino を使用した Iceberg テーブルの操作

このセクションでは、Amazon EMR で Trino を使用して Iceberg テーブルをセットアップおよび操作する方法について説明します。例は、EC2 クラスター上の Amazon EMR で実行できる定型コードです。このセクションのコード例と設定は、Amazon EMR リリース emr-7.9.0 を使用していることを前提としています。

EC2 での Amazon EMR のセットアップ

次の内容の iceberg.propertiesファイルを作成します。iceberg.file-format=parquet 設定は、 CREATE TABLE ステートメントで形式が明示的に指定されていない場合に、新しいテーブルのデフォルトのストレージ形式を決定します。
```
connector.name=iceberg
iceberg.catalog.type=glue
iceberg.file-format=parquet
fs.native-s3.enabled=true
```
iceberg.properties ファイルを S3 バケットにアップロードします。
S3 バケットからiceberg.propertiesファイルをコピーし、作成する Amazon EMR クラスターに Trino 設定ファイルとして保存するブートストラップアクションを作成します。を S3 バケット名<S3-bucket-name>に置き換えてください。
```
#!/bin/bash
set -ex
sudo aws s3 cp s3://<S3-bucket-name>/iceberg.properties /etc/trino/conf/catalog/iceberg.properties
```
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> インスタンスプロファイルを使用します。
Amazon EMR クラスターが初期化されたら、次のコマンドを実行して Trino セッションを初期化できます。
```
trino-cli
```
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 ステートメントを使用して、アップサートオペレーション (同時に新しいレコードを挿入し、既存のレコードを更新) を実行できます。例を示します。


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 テーブル内の書き込みオペレーション (挿入、更新、アップサート、または削除) ごとに、新しいスナップショットが作成されます。その後、これらのスナップショットをタイムトラベルに使用できます。さかのぼって、過去のテーブルのステータスを確認できます。

次のタイムトラベルクエリは、特定のに基づいてテーブルのステータスを表示します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-read設計に従うため、更新や削除の影響を受けるデータファイル全体を書き換えるのではなく、位置削除ファイルを作成します。copy-on-write方式を使用する場合は、書き込みオペレーションに Spark を使用することを検討してください。

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

Spark を使用した Iceberg テーブルの操作

Firehose を使用した Iceberg テーブルの操作