Arbeiten mit Iceberg-Tabellen mithilfe von Trino

In diesem Abschnitt wird beschrieben, wie Sie Iceberg-Tabellen mithilfe von Trino auf Amazon EMR einrichten und betreiben. Bei den Beispielen handelt es sich um Standardcode, den Sie auf einem Amazon EMR-Cluster ausführen können. EC2 Bei den Codebeispielen und Konfigurationen in diesem Abschnitt wird davon ausgegangen, dass Sie die Amazon EMR-Version emr-7.9.0 verwenden.

Amazon EMR bei der Einrichtung EC2

1. Erstellen Sie eine iceberg.properties Datei mit dem folgenden Inhalt. Die iceberg.file-format=parquet Einstellung bestimmt das Standardspeicherformat für neue Tabellen, wenn das Format in der CREATE TABLE Anweisung nicht explizit angegeben ist.

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

2. Laden Sie die Datei iceberg.properties in Ihren S3-Bucket hoch. 

3. Erstellen Sie eine Bootstrap-Aktion, die die iceberg.properties Datei aus Ihrem S3-Bucket kopiert und als Trino-Konfigurationsdatei auf dem Amazon EMR-Cluster speichert, den Sie erstellen werden. Stellen Sie sicher, dass Sie es durch Ihren <S3-bucket-name> S3-Bucket-Namen ersetzen. 

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

4. Erstellen Sie einen Amazon EMR-Cluster, auf dem Trino installiert ist, und geben Sie die Ausführung des vorherigen Skripts als Bootstrap-Aktion an. Hier ist ein Beispielbefehl AWS Command Line Interface (AWS CLI) zum Erstellen des Clusters:

   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>"}'

   wo Sie ersetzen:

   • <S3-bucket-name>mit Ihrem S3-Bucket-Namen

   • <region>mit deinem spezifischen AWS-Region

   • <key-name>mit deinem key pair. Wenn das key pair nicht existiert, wird es erstellt.

   • <IAM-service-role>mit Ihrer Amazon EMR-Servicerolle, die dem Prinzip der geringsten Rechte folgt. 

   • <EMR-EC2-instance-profile>mit Ihrem Instance-Profil. 

5. Wenn der Amazon EMR-Cluster initialisiert wurde, können Sie eine Trino-Sitzung initialisieren, indem Sie den folgenden Befehl ausführen:

   trino-cli

6. In der Trino CLI können Sie die Kataloge anzeigen, indem Sie Folgendes ausführen:

   SHOW CATALOGS;

Erstellen von Iceberg-Tabellen

Um eine Iceberg-Tabelle zu erstellen, können Sie die Anweisung verwenden. CREATE TABLE  Hier ist ein Beispiel für die Erstellung einer partitionierten Tabelle, die die versteckte Iceberg-Partitionierung verwendet:

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>');

Anmerkung

Wenn Sie das Format nicht angeben, wird der iceberg.file-format Wert verwendet, den Sie im vorherigen Abschnitt konfiguriert haben.

Verwenden Sie den INSERT INTO Befehl, um Daten einzufügen. Ein Beispiel:

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');

Aus Iceberg-Tabellen lesen

Sie können den aktuellen Status Ihrer Iceberg-Tabelle mithilfe einer SELECT Anweisung wie folgt ablesen:

SELECT * FROM iceberg.iceberg_db.iceberg_table;

Daten in Iceberg-Tabellen importieren

Sie können einen Upsert-Vorgang ausführen (gleichzeitig neue Datensätze einfügen und bestehende aktualisieren), indem Sie die Anweisung verwenden. MERGE INTO Ein Beispiel:

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);

Löschen von Datensätzen aus Iceberg-Tabellen

Um Daten aus einer Iceberg-Tabelle zu löschen, verwenden Sie den DELETE FROM Ausdruck und geben Sie einen Filter an, der den zu löschenden Zeilen entspricht. Ein Beispiel:

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

Iceberg-Tabellen-Metadaten abfragen

Iceberg bietet Zugriff auf seine Metadaten über SQL. Sie können auf die Metadaten für jede beliebige Tabelle (<table_name>) zugreifen, indem Sie den Namespace abfragen. "<table_name>.$<metadata_table>" Eine vollständige Liste der Metadatentabellen finden Sie in der Iceberg-Dokumentation unter Tabellen überprüfen.

Hier ist ein Beispiel für eine Liste von Abfragen zur Untersuchung von Iceberg-Metadaten:

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";

Beispielsweise würde diese Abfrage:

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

liefert die Ausgabe:

Zeitreisen nutzen

Jeder Schreibvorgang (Einfügen, Aktualisieren, Hochsetzen oder Löschen) in einer Iceberg-Tabelle erstellt einen neuen Snapshot. Sie können diese Snapshots dann für Zeitreisen verwenden, um in die Vergangenheit zu reisen und den Status einer Tabelle in der Vergangenheit zu überprüfen.

Die folgende Zeitreise-Abfrage zeigt den Status einer Tabelle auf der Grundlage eines bestimmten Status an: snapshot_id

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

Die folgende Zeitreiseabfrage zeigt den Status einer Tabelle auf der Grundlage eines bestimmten Zeitstempels an:

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

Überlegungen zur Verwendung von Iceberg mit Trino

Die Schreiboperationen von Trino auf Iceberg-Tabellen folgen dem merge-on-readDesign, sodass sie positionsabhängige Löschdateien erstellen, anstatt ganze Datendateien neu zu schreiben, die von Aktualisierungen oder Löschungen betroffen sind. Wenn Sie diesen copy-on-write Ansatz verwenden möchten, sollten Sie die Verwendung von Spark für Schreiboperationen in Betracht ziehen.