Trabalhando com tabelas Iceberg usando o Trino

Esta seção descreve como configurar e operar tabelas Iceberg usando o Trino no Amazon EMR. Os exemplos são códigos padronizados que você pode executar em um Amazon EMR em um cluster. EC2 Os exemplos de código e as configurações nesta seção pressupõem que você esteja usando o Amazon EMR versão emr-7.9.0.

Amazon EMR na configuração EC2

1. Crie um iceberg.properties arquivo com o conteúdo a seguir. A iceberg.file-format=parquet configuração determina o formato de armazenamento padrão para novas tabelas se o formato não for especificado explicitamente na CREATE TABLE instrução.

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

2. Faça upload do arquivo iceberg.properties no bucket do S3. 

3. Crie uma ação de bootstrap que copie o iceberg.properties arquivo do seu bucket do S3 e o armazene como um arquivo de configuração Trino no cluster do Amazon EMR que você criará. Certifique-se de <S3-bucket-name> substituir pelo nome do bucket do S3. 

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

4. Crie um cluster do Amazon EMR com o Trino instalado e especifique a execução do script anterior como uma ação de bootstrap. Aqui está um exemplo de comando AWS Command Line Interface (AWS CLI) para criar o cluster:

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

   onde você substitui:

   • <S3-bucket-name>com o nome do seu bucket S3

   • <region>com seu específico Região da AWS

   • <key-name>com seu key pair. Se o key pair não existir, ele será criado.

   • <IAM-service-role>com sua função de serviço do Amazon EMR que segue o princípio do menor privilégio. 

   • <EMR-EC2-instance-profile>com seu perfil de instância. 

5. Quando o cluster do Amazon EMR for inicializado, você poderá inicializar uma sessão do Trino executando o seguinte comando:

   trino-cli

6. Na CLI do Trino, você pode visualizar os catálogos executando:

   SHOW CATALOGS;

Criar tabelas Iceberg

Para criar uma tabela Iceberg, você pode usar a CREATE TABLE declaração.  Aqui está um exemplo de criação de uma tabela particionada que usa o particionamento oculto do 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

Se você não especificar o formato, o iceberg.file-format valor que você configurou na seção anterior será usado.

Para inserir dados, use o INSERT INTO comando. Veja um exemplo abaixo:

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

Lendo as tabelas do Iceberg

Você pode ler o status mais recente da sua tabela Iceberg usando uma SELECT declaração, da seguinte forma:

SELECT * FROM iceberg.iceberg_db.iceberg_table;

Inserindo dados em tabelas Iceberg

Você pode executar uma operação de upsert (inserir simultaneamente novos registros e atualizar os existentes) usando a MERGE INTO instrução. Veja um exemplo abaixo:

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

Excluindo registros das tabelas do Iceberg

Para excluir dados de uma tabela do Iceberg, use a DELETE FROM expressão e especifique um filtro que corresponda às linhas a serem excluídas. Veja um exemplo abaixo:

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

Consultar metadados de tabela do Iceberg

O Iceberg fornece acesso aos seus metadados por meio de SQL. Você pode acessar os metadados de qualquer tabela (<table_name>) consultando o namespace. "<table_name>.$<metadata_table>" Para obter uma lista completa das tabelas de metadados, consulte Como inspecionar tabelas na documentação do Iceberg.

Aqui está um exemplo de lista de consultas para inspecionar os metadados do 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 exemplo, esta consulta:

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

fornece a saída:

Usando a viagem no tempo

Cada operação de gravação (inserção, atualização, atualização ou exclusão) em uma tabela do Iceberg cria um novo instantâneo. Em seguida, você pode usar esses instantâneos para viajar no tempo, para voltar no tempo e verificar o status de uma tabela no passado.

A consulta de viagem no tempo a seguir exibe o status de uma tabela com base em uma tabela específicasnapshot_id:

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

A consulta de viagem no tempo a seguir exibe o status de uma tabela com base em um timestamp específico:

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

Considerações ao usar o Iceberg com Trino

As operações de gravação do Trino nas tabelas Iceberg seguem o merge-on-readdesign, então elas criam arquivos de exclusão posicional em vez de reescrever arquivos de dados inteiros que são afetados por atualizações ou exclusões. Se você quiser usar a copy-on-write abordagem, considere usar o Spark para operações de gravação.