Trabajar con tablas Iceberg mediante Trino

En esta sección se describe cómo configurar y operar tablas Iceberg mediante Trino en Amazon EMR. Los ejemplos son código repetitivo que puede ejecutar en un clúster de Amazon EMR. EC2 En los ejemplos de código y las configuraciones de esta sección se supone que está utilizando la versión emr-7.9.0 de Amazon EMR.

Amazon EMR durante la configuración EC2

1. Cree un iceberg.properties archivo con el siguiente contenido. La iceberg.file-format=parquet configuración determina el formato de almacenamiento predeterminado para las tablas nuevas si el formato no se especifica explícitamente en la CREATE TABLE instrucción.

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

2. Cargue el archivo iceberg.properties en el bucket de S3. 

3. Cree una acción de arranque que copie el iceberg.properties archivo de su bucket de S3 y lo almacene como un archivo de configuración de Trino en el clúster de Amazon EMR que va a crear. Asegúrese de reemplazarlo por el nombre <S3-bucket-name> de su bucket de S3. 

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

4. Cree un clúster de Amazon EMR con Trino instalado y especifique la ejecución del script anterior como una acción de arranque. Este es un ejemplo de comando AWS Command Line Interface (AWS CLI) para crear el clúster:

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

   donde sustituyes:

   • <S3-bucket-name>con el nombre de su bucket de S3

   • <region>con tu específico Región de AWS

   • <key-name>con tu key pair. Si el par de claves no existe, se creará.

   • <IAM-service-role>con su función de servicio de Amazon EMR que siga el principio del privilegio mínimo. 

   • <EMR-EC2-instance-profile>con su perfil de instancia. 

5. Cuando se haya inicializado el clúster de Amazon EMR, puede inicializar una sesión de Trino ejecutando el siguiente comando:

   trino-cli

6. En la CLI de Trino, puede ver los catálogos ejecutando:

   SHOW CATALOGS;

Creación de tablas de Iceberg

Para crear una tabla de iceberg, puede usar la sentencia. CREATE TABLE  Este es un ejemplo de cómo crear una tabla particionada que usa la partición oculta de 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>');

nota

Si no especificas el formato, se usará el iceberg.file-format valor que configuraste en la sección anterior.

Para insertar datos, utilice el INSERT INTO comando. A continuación se muestra un ejemplo:

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

Lectura de tablas de iceberg

Puede leer el estado más reciente de su tabla de iceberg mediante una SELECT declaración, de la siguiente manera:

SELECT * FROM iceberg.iceberg_db.iceberg_table;

Dividir los datos en tablas de iceberg

Puede realizar una operación de alteración (insertar simultáneamente nuevos registros y actualizar los existentes) mediante la instrucción. MERGE INTO A continuación se muestra un ejemplo:

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

Eliminar registros de las tablas de Iceberg

Para eliminar datos de una tabla de iceberg, utilice la DELETE FROM expresión y especifique un filtro que coincida con las filas que desee eliminar. A continuación se muestra un ejemplo:

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

Consulta de los metadatos de la tabla de Iceberg

Iceberg proporciona acceso a sus metadatos a través de SQL. Puede acceder a los metadatos de cualquier tabla (<table_name>) consultando el espacio de nombres. "<table_name>.$<metadata_table>" Para obtener una lista completa de las tablas de metadatos, consulte Inspección de tablas en la documentación de Iceberg.

A continuación, se muestra un ejemplo de una lista de consultas para inspeccionar los metadatos de 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";

Por ejemplo, esta consulta:

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

proporciona el resultado:

Uso del viaje en el tiempo

Cada operación de escritura (insertar, actualizar, modificar o eliminar) en una tabla de Iceberg crea una nueva instantánea. A continuación, puede utilizar estas instantáneas para viajar en el tiempo, es decir, para retroceder en el tiempo y comprobar el estado de una tabla en el pasado.

La siguiente consulta de viaje en el tiempo muestra el estado de una tabla en función de un dato específico: snapshot_id

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

La siguiente consulta de viaje en el tiempo muestra el estado de una tabla en función de una marca de tiempo específica:

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

Consideraciones a la hora de utilizar Iceberg con Trino

Las operaciones de escritura de Trino en las tablas de Iceberg siguen el merge-on-readdiseño, por lo que crean archivos de eliminación posicional en lugar de reescribir archivos de datos completos que se ven afectados por las actualizaciones o eliminaciones. Si quieres usar copy-on-write este enfoque, considera usar Spark para las operaciones de escritura.