E-MapReduce:Iceberg connector

Prerequisites

Before you begin, ensure that you have:

• A DataLake cluster or Hadoop cluster with the Presto service enabled. See Create a cluster.

Limitations

• Only DataLake clusters and Hadoop clusters running EMR V3.38.0 or later support the Iceberg connector.

• If you selected DLF Unified Metadata when creating the cluster, you cannot write data to Iceberg tables.

Configure the Iceberg connector

For the general procedure to modify connector configurations, see Configure connectors.

Query Iceberg tables

The following steps show how to create a schema and table, insert data, and query results using standard Trino SQL.

SQL syntax

The Iceberg connector supports reading and writing data and metadata in Iceberg tables. In addition to standard SQL, it supports the following statements:

Table properties

Use the WITH clause to set the following properties when creating an Iceberg table:

Example:

CREATE TABLE test_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE)
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['c1', 'c2'],
    location = '/var/my_tables/test_table');

Partition a table

The Iceberg connector supports function-based partitioning. Use the following functions in the partitioning property:

Example: Partition the customer_orders table by order month, account number hash (10 buckets), and country:

CREATE TABLE iceberg.testdb.customer_orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (partitioning = ARRAY['month(order_date)', 'bucket(account_number, 10)', 'country']);

Delete data by partition

For a partitioned table, if you include a WHERE clause in a DELETE statement to filter partitions, the Iceberg connector deletes the partitions that match the filter conditions. For example, the following statement deletes all partitions where country = 'US' from the customer_orders table:

DELETE FROM iceberg.testdb.customer_orders
WHERE country = 'US';

You can use the Iceberg connector to delete data only by partition. The following statement fails to be executed because the WHERE clause filters specific rows within partitions rather than complete partitions:

DELETE FROM iceberg.testdb.customer_orders
WHERE country = 'US' AND customer = 'Freds Foods';

Query system tables

The Iceberg connector exposes system tables that provide metadata about each Iceberg table.

Query partitions — includes the minimum and maximum values for each partition key column:

SELECT * FROM iceberg.testdb."customer_orders$partitions";

Query snapshots — lists all snapshots with their commit timestamps:

SELECT * FROM iceberg.testdb."customer_orders$snapshots"
ORDER BY committed_at DESC;

Roll back to a snapshot

Iceberg tables support snapshots. The Iceberg connector provides a system snapshot table for each Iceberg table. Snapshot IDs are of the BIGINT data type.

To revert a table to a previous state, first get the target snapshot ID, then call the rollback procedure.

1. Get the latest snapshot ID:

   SELECT snapshot_id
   FROM iceberg.testdb."customer_orders$snapshots"
   ORDER BY committed_at DESC
   LIMIT 1;

2. Roll back to the snapshot:

   CALL iceberg.system.rollback_to_snapshot('testdb', 'customer_orders', 895459706749342****);

   Replace 895459706749342**** with the actual snapshot ID (a BIGINT value).

Manage materialized views

A materialized view in the Iceberg connector consists of a view definition and a backing Iceberg table. The table name is stored as a materialized view property; data is stored in the Iceberg table.

What's next

• Overview — learn about Iceberg concepts including snapshots, partitions, and table format versions.

• Configure connectors — general guide for modifying connector configurations in EMR.