Use Iceberg - E-MapReduce - Alibaba Cloud Documentation Center

Iceberg is an open table format for a data lake. You can use Iceberg to quickly build a data lake storage service on Hadoop Distributed File System (HDFS) or Alibaba Cloud Object Storage Service (OSS). This topic describes how to read from and write to Iceberg tables in EMR Serverless Spark.

Prerequisites

A workspace is created. Create a workspace.

Procedure

Note

Both SparkSQL and notebooks support reading from and writing to Iceberg tables. This topic uses a SparkSQL job as an example.

Step 1: Create a session resource

Go to the Sessions page.
1. Log on to the EMR console.
2. In the left-side navigation pane, choose EMR Serverless > Spark.
3. On the Spark page, click the name of the target workspace.
4. On the EMR Serverless Spark page, click Sessions in the left-side navigation pane.
On the SQL Session page, click Connect to SQL Session.

On the Connect to SQL Session page, configure the following parameters in the Spark Configuration section, and then click Create. For more information, see Manage SQL sessions.

Spark reads from and writes to Iceberg tables using a catalog. You can choose a catalog based on your use case. For more information about catalogs, see Manage data catalogs.

Data catalog

If you use a data catalog, you do not need to configure parameters in the session. On the Catalog page, click Add Catalog. You can then select the data catalog directly in your SparkSQL development environment.

Note

To access Iceberg in DLF (formerly DLF 2.5), use engine version esr-4.7.0, esr-3.6.0, or later.
To access Iceberg in DLF 1.0 (formerly DLF-Legacy) or Hive Metastore, use engine version esr-4.3.0, esr-3.3.0, esr-2.7.0, or later.

Custom catalog

DLF (formerly DLF 2.5)

Note

Engine version requirement: esr-4.7.0, esr-3.6.0, or later.

spark.sql.extensions org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.iceberg_catalog org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg_catalog.catalog-impl org.apache.iceberg.rest.RESTCatalog
spark.sql.catalog.iceberg_catalog.uri http://<regionID>-vpc.dlf.aliyuncs.com
spark.sql.catalog.iceberg_catalog.warehouse  <catalog_name>
spark.sql.catalog.iceberg_catalog.io-impl org.apache.iceberg.rest.DlfFileIO
spark.sql.catalog.iceberg_catalog.rest.auth.type sigv4
spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type none
spark.sql.catalog.iceberg_catalog.rest.signing-region <regionID>
spark.sql.catalog.iceberg_catalog.rest.signing-name DlfNext
spark.sql.catalog.iceberg_catalog.rest.access-key-id <access_key_id>
spark.sql.catalog.iceberg_catalog.rest.secret-access-key <access_key_secret>

The following table describes the parameters.

Parameter	Description	Example value
`spark.sql.extensions`	Enables Iceberg Spark extensions.	Fixed value: `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.iceberg_catalog`	Registers a Spark catalog named `iceberg_catalog`.	Fixed value: `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.iceberg_catalog.catalog-impl`	Specifies the underlying catalog implementation as the Iceberg REST catalog.	`org.apache.iceberg.rest.RESTCatalog`
`spark.sql.catalog.iceberg_catalog.uri`	The REST API endpoint for the DLF Iceberg service. The format is `http://<regionID>-vpc.dlf.aliyuncs.com`.	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.iceberg_catalog.warehouse`	The name of the associated DLF catalog. Note Do not associate a DLF catalog that is created from a data share.	`<catalog_name>`
`spark.sql.catalog.iceberg_catalog.io-impl`	Uses the DLF-customized FileIO implementation.	Fixed value: `org.apache.iceberg.rest.DlfFileIO`
`spark.sql.catalog.iceberg_catalog.rest.auth.type`	Enables the AWS Signature Version 4 (SigV4) signing mechanism for authenticating REST requests.	`sigv4`
`spark.sql.catalog.iceberg_catalog.rest.auth.sigv4.delegate-auth-type`	Disables delegated authentication. The client must provide an AccessKey ID and AccessKey Secret to sign requests.	`none`
`spark.sql.catalog.iceberg_catalog.rest.signing-region`	The region used for signing. This value must match the region where the DLF service is deployed.	`cn-hangzhou`
`spark.sql.catalog.iceberg_catalog.rest.signing-name`	The service name used for signing.	Fixed value: `DlfNext`
`spark.sql.catalog.iceberg_catalog.rest.access-key-id`	The AccessKey ID of your Alibaba Cloud account or RAM user.	`<access_key_id>`
`spark.sql.catalog.iceberg_catalog.rest.secret-access-key`	The AccessKey Secret of your Alibaba Cloud account or RAM user.	`<access_key_secret>`

DLF 1.0 (formerly DLF-Legacy)

Note

Engine version requirement: esr-4.3.0, esr-3.3.0, esr-2.7.0, or later.

Metadata is stored in DLF 1.0 (formerly DLF-Legacy).

spark.sql.extensions                          org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<catalogName>               org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.<catalogName>.catalog-impl  org.apache.iceberg.aliyun.dlf.hive.DlfCatalog
spark.sql.catalog.<catalogName>.dlf.catalog.id <catalog_name>

The following table describes the parameters.

Parameter	Description	Example value
`spark.sql.extensions`	Enables Iceberg Spark extensions.	Fixed value: `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.<catalogName>`	Registers a catalog named `<catalogName>`.	Fixed value: `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.catalog-impl`	Specifies the Alibaba Cloud DLF 1.0 Hive-compatible implementation, which connects to the DLF 1.0 metadata service.	Fixed value: `org.apache.iceberg.aliyun.dlf.hive.DlfCatalog`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	Specifies the name of the associated DLF catalog.	`<catalog_name>`

Hive Metastore

Metadata is stored in a specified Hive Metastore.

spark.sql.extensions                          org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<catalogName>               org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.<catalogName>.catalog-impl  org.apache.iceberg.hive.HiveCatalog
spark.sql.catalog.<catalogName>.uri           thrift://<yourHMSUri>:<port>

The following table describes the parameters.

Parameter	Description	Example value
`spark.sql.extensions`	Enables Iceberg Spark extensions.	Fixed value: `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.<catalogName>`	Registers a catalog named `<catalogName>`.	Fixed value: `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.catalog-impl`	Specifies that the catalog uses the official Iceberg `HiveCatalog` implementation to manage metadata in the Hive Metastore.	Fixed value: `org.apache.iceberg.hive.HiveCatalog`
`spark.sql.catalog.<catalogName>.uri`	The URI of the Hive Metastore. The format is `thrift://<Hive metastore IP address>:9083`. `<Hive metastore IP address>` is the internal IP address of the HMS service. If you need to specify an external metastore service, see Connect to an external Hive Metastore Service.	`thrift://192.168..:9083`

File system

Metadata is stored in a file system.

spark.sql.extensions                          org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<catalogName>               org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.<catalogName>.type          hadoop
spark.sql.catalog.<catalogName>.warehouse     oss://<yourBucketName>/warehouse

The following table describes the parameters.

Parameter	Description	Example value
`spark.sql.extensions`	Enables Iceberg Spark extensions.	Fixed value: `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions`
`spark.sql.catalog.<catalogName>`	Registers a catalog named `<catalogName>`.	Fixed value: `org.apache.iceberg.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.type`	Specify the Catalog type as `hadoop`. This means that HadoopCatalog stores metadata directly in the file system and does not require a Hive Metastore.	`hadoop`
`spark.sql.catalog.<catalogName>.warehouse`	Specifies the path for metadata storage. In this path, `<yourBucketName>` represents the name of the OSS bucket.	`oss://<yourBucketName>/warehouse`

Step 2: Read and write Iceberg tables

Go to the Data Development page.

On the EMR Serverless Spark page, click Development in the left-side navigation pane.
On the Development tab, click the icon.
In the Create dialog box, enter a name such as users_task, set the Type to the default SparkSQL, and then click OK.

Copy the following code into the new SparkSQL tab (users_task).

Note

If you do not specify a database, the table is created in the default database of the catalog. You can also create and specify a different database.

-- Create a database.
CREATE DATABASE IF NOT EXISTS iceberg_catalog.db;
-- Create a non-partitioned table.
CREATE TABLE iceberg_catalog.db.tbl (
    id BIGINT NOT NULL COMMENT 'unique id',
    data STRING
)
USING iceberg;
-- Insert data into the non-partitioned table.
INSERT INTO iceberg_catalog.db.tbl VALUES
(1, 'Alice'),
(2, 'Bob'),
(3, 'Charlie');
-- Query all data from the non-partitioned table.
SELECT * FROM iceberg_catalog.db.tbl;
-- Query the non-partitioned table with a condition.
SELECT * FROM iceberg_catalog.db.tbl WHERE id = 2;
-- Update data in the non-partitioned table.
UPDATE iceberg_catalog.db.tbl SET data = 'David' WHERE id = 3;
-- Query again to confirm the update.
SELECT * FROM iceberg_catalog.db.tbl WHERE id = 3;
-- Delete data from the non-partitioned table.
DELETE FROM iceberg_catalog.db.tbl WHERE id = 1;
-- Query again to confirm the deletion.
SELECT * FROM iceberg_catalog.db.tbl;
-- Create a partitioned table.
CREATE TABLE iceberg_catalog.db.part_tbl (
    id BIGINT,
    data STRING,
    category STRING,
    ts TIMESTAMP,
    dt DATE
)
USING iceberg
PARTITIONED BY (dt, category);
-- Insert data into the partitioned table.
INSERT INTO iceberg_catalog.db.part_tbl VALUES
  (1 , 'data-01', 'A', timestamp'2026-01-01 10:00:00', date'2026-01-01'),
  (2 , 'data-02', 'A', timestamp'2026-01-01 11:00:00', date'2026-01-01'),
  (3 , 'data-03', 'A', timestamp'2026-01-02 09:30:00', date'2026-01-02'),
  (4 , 'data-04', 'B', timestamp'2026-01-02 12:15:00', date'2026-01-02'),
  (5 , 'data-05', 'B', timestamp'2026-01-03 08:05:00', date'2026-01-03'),
  (6 , 'data-06', 'B', timestamp'2026-01-03 14:20:00', date'2026-01-03'),
  (7 , 'data-07', 'C', timestamp'2026-01-04 16:45:00', date'2026-01-04'),
  (8 , 'data-08', 'C', timestamp'2026-01-04 18:10:00', date'2026-01-04'),
  (9 , 'data-09', 'C', timestamp'2026-01-05 07:55:00', date'2026-01-05'),
  (10, 'data-10', 'A', timestamp'2026-01-05 13:35:00', date'2026-01-05');
-- Query all data from the partitioned table.
SELECT * FROM iceberg_catalog.db.part_tbl;
-- Query data from the partition where dt='2026-01-01'.
SELECT * FROM iceberg_catalog.db.part_tbl WHERE dt='2026-01-01';
-- Query data for a specific category.
SELECT * FROM iceberg_catalog.db.part_tbl WHERE category = 'A';
-- Perform a query with multiple conditions (day + category).
SELECT * FROM iceberg_catalog.db.part_tbl 
WHERE dt='2026-01-01'
  AND category = 'A';
-- Aggregate and count the number of records in each category.
SELECT category, COUNT(*) AS count 
FROM iceberg_catalog.db.part_tbl 
GROUP BY category;
-- Drop the database (use with caution). Before you drop the database, make sure that it contains no tables.
-- DROP DATABASE iceberg_catalog.db;

In the session drop-down list, select the SQL session instance that you created, and then click Run. After the job completes, the results are displayed below. After you run the aggregation query, the output is as follows: category B (3), category C (3), and category A (4).

References

Quick Start for SparkSQL development
For more information about using and configuring Iceberg, see Apache Iceberg.
For details on creating an SQL session resource, see Manage SQL sessions.
For details on creating a notebook session resource, see Manage notebook sessions.