GuideConnect engines and clients

Use Iceberg REST clients with Unity Catalog

Connect Spark, DuckDB, and PyIceberg to Unity Catalog's Iceberg REST API, then create, write, change, rename, and drop managed Iceberg tables with vended S3 credentials.

Unity Catalog serves the Iceberg REST catalog API, so any client that speaks it can create, write, read, rename, and drop Iceberg tables in a Unity Catalog catalog. The server allocates storage for each new table and vends short-lived credentials for it, so clients never hold bucket keys. Iceberg tables live next to Delta tables in the same schemas, under the same privileges.

How Iceberg names map to the catalog

Every client is configured with the same three values:

Iceberg REST setting	Value
Catalog URI	`http://localhost:8080/api/2.1/unity-catalog/iceberg`
Warehouse	The name of one Unity Catalog catalog, here `retail`. A client works with one catalog at a time; configure a second client catalog for another.
Token	A Unity Catalog access token, sent as a bearer token. The local server has authorization disabled and accepts any value.

Inside that catalog, an Iceberg namespace is a Unity Catalog schema, and an Iceberg table is a table whose data source format is ICEBERG. Namespaces don't nest. A table created without a location is a managed table under the schema's or catalog's storage root; the server records where it lives and deletes its files when you drop it.

Through this API, the server lists Iceberg tables, and Delta tables only when they carry UniForm Iceberg metadata. Reach other Delta tables with a Delta client, or with both in one Spark session.

Connect a client

Open a Python session with PySpark:

uv run --with pyspark==4.1.0 python

spark_iceberg.pyFull script

from pyspark.sql import SparkSession

ICEBERG_SPARK = "org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:1.11.0"
ICEBERG_AWS = "org.apache.iceberg:iceberg-aws-bundle:1.11.0"

spark = (
    SparkSession.builder.appName("uc-iceberg")
    .config("spark.jars.packages", f"{ICEBERG_SPARK},{ICEBERG_AWS}")
    .config(
        "spark.sql.extensions",
        "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
    )
    .config("spark.sql.catalog.retail", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.retail.type", "rest")
    .config(
        "spark.sql.catalog.retail.uri",
        "http://localhost:8080/api/2.1/unity-catalog/iceberg",
    )
    .config("spark.sql.catalog.retail.warehouse", "retail")
    .config("spark.sql.catalog.retail.token", "not-used")
    .config("spark.sql.defaultCatalog", "retail")
    .getOrCreate()
)

This is Iceberg's own Spark catalog, not the Unity Catalog connector. The Iceberg runtime is built per Spark minor version, and the AWS bundle adds the S3 file system that reads and writes table files with the credentials the server vends. The warehouse setting picks the Unity Catalog catalog. The retail in spark.sql.catalog.retail is only Spark's name for it, and making it the default catalog lets later statements name tables as sales.orders.

Open a Python session with DuckDB:

uv run --with duckdb==1.5.6 python

Install and load the iceberg extension:

duckdb_iceberg.pyFull script

import duckdb

con = duckdb.connect()
con.execute("INSTALL iceberg")
con.execute("LOAD iceberg")
con.execute("SET iceberg_use_metadata_log = false")

Then create a secret that holds the token and attach the catalog:

duckdb_iceberg.pyFull script

con.execute("CREATE SECRET uc (TYPE iceberg, TOKEN 'not-used')")
con.execute("""
    ATTACH 'retail' AS retail (
        TYPE iceberg,
        SECRET uc,
        ENDPOINT 'http://localhost:8080/api/2.1/unity-catalog/iceberg'
    )
""")

ATTACH takes the warehouse, here retail. DuckDB uses the credentials the server vends for each table.

Turning off iceberg_use_metadata_log avoids an intermittent commit failure on a session's second write to a table.

Note

DuckDB doesn't read AWS_ENDPOINT_URL. Against the local environment's simulated S3, point it at the endpoint yourself:

duckdb_iceberg.pyFull script

con.execute("SET s3_endpoint = 'localhost:9000'")
con.execute("SET s3_use_ssl = false")
con.execute("SET s3_url_style = 'path'")

On real AWS, leave these settings out.

Open a Python session with PyIceberg:

uv run --with "pyiceberg[pyarrow]==0.12.0" python

pyiceberg_tables.pyFull script

import pyarrow as pa
from pyiceberg.catalog import load_catalog

pyiceberg_tables.pyFull script

catalog = load_catalog(
    "retail",
    type="rest",
    uri="http://localhost:8080/api/2.1/unity-catalog/iceberg",
    warehouse="retail",
    token="not-used",
)

load_catalog takes the client-side name first, then the same three values. PyIceberg uses the credentials the server vends for each table.

Create a schema

spark_iceberg.pyFull script

spark.sql("CREATE NAMESPACE staging")
spark.sql("SHOW NAMESPACES").show()

duckdb_iceberg.pyFull script

con.execute("CREATE SCHEMA retail.staging")
con.sql(
    "SELECT schema_name FROM duckdb_schemas() WHERE database_name = 'retail'"
).show()

pyiceberg_tables.pyFull script

catalog.create_namespace("staging")
print(catalog.list_namespaces())

The new namespace is an ordinary Unity Catalog schema, retail.staging, and other clients see it at once.

Create a table

spark_iceberg.pyFull script

spark.sql("""
    CREATE TABLE sales.orders (
        order_id BIGINT,
        customer STRING,
        amount DOUBLE
    ) USING iceberg
""")

duckdb_iceberg.pyFull script

con.execute("""
    CREATE TABLE retail.sales.orders (
        order_id BIGINT,
        customer VARCHAR,
        amount DOUBLE
    )
""")

pyiceberg_tables.pyFull script

orders_schema = pa.schema(
    [("order_id", pa.int64()), ("customer", pa.string()), ("amount", pa.float64())]
)
orders = catalog.create_table("sales.orders", schema=orders_schema)

The server creates the table as MANAGED, with data source format ICEBERG, at a location under the catalog's storage root: s3://uc-docs/retail/managed/__unitystorage/catalogs/<catalog id>/tables/<table id>.

Write and read rows

spark_iceberg.pyFull script

spark.sql("""
    INSERT INTO sales.orders VALUES
        (1, 'ada', 9.5),
        (2, 'grace', 20.0),
        (3, 'linus', 3.25)
""")

spark_iceberg.pyFull script

spark.sql("SELECT * FROM sales.orders WHERE amount > 5 ORDER BY order_id").show()

+--------+--------+------+
|order_id|customer|amount|
+--------+--------+------+
|       1|     ada|   9.5|
|       2|   grace|  20.0|
+--------+--------+------+

duckdb_iceberg.pyFull script

con.execute("""
    INSERT INTO retail.sales.orders VALUES
        (1, 'ada', 9.5),
        (2, 'grace', 20.0),
        (3, 'linus', 3.25)
""")

duckdb_iceberg.pyFull script

con.sql("SELECT * FROM retail.sales.orders WHERE amount > 5 ORDER BY order_id").show()

┌──────────┬──────────┬────────┐
│ order_id │ customer │ amount │
│  int64   │ varchar  │ double │
├──────────┼──────────┼────────┤
│        1 │ ada      │    9.5 │
│        2 │ grace    │   20.0 │
└──────────┴──────────┴────────┘

pyiceberg_tables.pyFull script

orders.append(
    pa.table(
        {
            "order_id": [1, 2, 3],
            "customer": ["ada", "grace", "linus"],
            "amount": [9.5, 20.0, 3.25],
        },
        schema=orders_schema,
    )
)

pyiceberg_tables.pyFull script

print(orders.scan(row_filter="amount > 5").to_arrow().sort_by("order_id"))

pyarrow.Table
order_id: int64
customer: string
amount: double
----
order_id: [[1,2]]
customer: [["ada","grace"]]
amount: [[9.5,20]]

The client writes data and metadata files with the vended credentials, then asks the server to commit the new snapshot. The server accepts a commit only if the table hasn't changed since the client read it, so concurrent writers can't overwrite each other.

Update and delete rows

spark_iceberg.pyFull script

spark.sql("UPDATE sales.orders SET amount = 21.0 WHERE order_id = 2")
spark.sql("DELETE FROM sales.orders WHERE customer = 'linus'")
spark.sql("""
    MERGE INTO sales.orders AS t
    USING (SELECT 4 AS order_id, 'barbara' AS customer, 12.0 AS amount) AS s
    ON t.order_id = s.order_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

UPDATE, DELETE FROM, and MERGE INTO need the Iceberg SQL extensions from the session configuration.

duckdb_iceberg.pyFull script

con.execute("UPDATE retail.sales.orders SET amount = 21.0 WHERE order_id = 2")
con.execute("DELETE FROM retail.sales.orders WHERE customer = 'linus'")

pyiceberg_tables.pyFull script

orders.delete(delete_filter="customer = 'linus'")
orders.overwrite(
    pa.table(
        {"order_id": [2], "customer": ["grace"], "amount": [21.0]},
        schema=orders_schema,
    ),
    overwrite_filter="order_id = 2",
)

PyIceberg has no UPDATE. Replace the matching rows with overwrite and an overwrite_filter.

Rename and drop a table

spark_iceberg.pyFull script

spark.sql("ALTER TABLE sales.orders RENAME TO sales.order_history")

spark_iceberg.pyFull script

spark.sql("DROP TABLE sales.order_history")
spark.sql("DROP NAMESPACE staging")

duckdb_iceberg.pyFull script

con.execute("ALTER TABLE retail.sales.orders RENAME TO order_history")

duckdb_iceberg.pyFull script

con.execute("DROP TABLE retail.sales.order_history")
con.execute("DROP SCHEMA retail.staging")

pyiceberg_tables.pyFull script

catalog.rename_table("sales.orders", "sales.order_history")

pyiceberg_tables.pyFull script

catalog.drop_table("sales.order_history")
catalog.drop_namespace("staging")

A rename keeps the table in its schema. Dropping a managed table removes it from the catalog, and a background worker on the server deletes its files. An external table, one created with an explicit location, keeps its files; the server accepts a purge request for it but leaves the files in place.

Only tables created through Iceberg REST can be dropped through it. Drop a Delta table through Spark's Unity Catalog connector or the Unity Catalog API.

Join Delta and Iceberg tables in Spark

One Spark session can hold both kinds of table: register the catalog twice, once through the Unity Catalog connector for Delta tables and once through Iceberg REST for Iceberg tables.

spark_cross_format.pyFull script

from pyspark.sql import SparkSession

PACKAGES = [
    "io.unitycatalog:unitycatalog-spark_4.1_2.13:0.7.0",
    "io.delta:delta-spark_4.1_2.13:4.3.1",
    "org.apache.hadoop:hadoop-aws:3.4.2",
    "org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:1.11.0",
    "org.apache.iceberg:iceberg-aws-bundle:1.11.0",
]

spark = (
    SparkSession.builder.appName("uc-delta-and-iceberg")
    .config("spark.jars.packages", ",".join(PACKAGES))
    .config(
        "spark.sql.extensions",
        "io.delta.sql.DeltaSparkSessionExtension,"
        "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
    )
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
    .config("spark.hadoop.fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
    # Delta tables, through the Unity Catalog connector.
    .config("spark.sql.catalog.retail", "io.unitycatalog.spark.UCSingleCatalog")
    .config("spark.sql.catalog.retail.uri", "http://localhost:8080")
    .config("spark.sql.catalog.retail.auth.type", "static")
    .config("spark.sql.catalog.retail.auth.token", "not-used")
    # Iceberg tables in the same catalog, through Iceberg REST.
    .config("spark.sql.catalog.retail_iceberg", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.retail_iceberg.type", "rest")
    .config(
        "spark.sql.catalog.retail_iceberg.uri",
        "http://localhost:8080/api/2.1/unity-catalog/iceberg",
    )
    .config("spark.sql.catalog.retail_iceberg.warehouse", "retail")
    .config("spark.sql.catalog.retail_iceberg.token", "not-used")
    .getOrCreate()
)

The connector needs hadoop-aws for s3:// paths; use the version that matches your Spark's Hadoop libraries (3.4.2 for Spark 4.1). For the connector settings, see Use Spark with Unity Catalog. Create a table of each kind, then query them together:

spark_cross_format.pyFull script

spark.sql("CREATE TABLE retail.sales.customers (id BIGINT, name STRING) USING DELTA")
spark.sql("INSERT INTO retail.sales.customers VALUES (1, 'ada'), (2, 'grace')")

spark.sql("""
    CREATE TABLE retail_iceberg.sales.orders (
        order_id BIGINT, customer_id BIGINT, amount DOUBLE
    ) USING iceberg
""")
spark.sql("""
    INSERT INTO retail_iceberg.sales.orders VALUES
        (10, 1, 9.5), (11, 2, 20.0), (12, 1, 1.5)
""")

spark_cross_format.pyFull script

spark.sql("""
    SELECT c.name, sum(o.amount) AS total
    FROM retail.sales.customers AS c
    JOIN retail_iceberg.sales.orders AS o ON o.customer_id = c.id
    GROUP BY c.name
    ORDER BY c.name
""").show()

+-----+-----+
| name|total|
+-----+-----+
|  ada| 11.0|
|grace| 20.0|
+-----+-----+

Name each table through the catalog for its format. The connector lists the Iceberg table too, but can't read it, and Iceberg REST doesn't list Delta tables.

Required privileges

With authorization enabled, the server checks the same privileges as for any other table. Reading needs USE CATALOG on the catalog, USE SCHEMA on the schema, and SELECT on the table. Writing also needs MODIFY; without it, the server vends read-only credentials. Creating a table needs CREATE TABLE on the schema, and creating a schema needs CREATE SCHEMA on the catalog. Owners have all of these. See Core concepts.

Next steps

Credential vending explains how the server scopes the credentials it hands each client.
External tables and catalog-managed Delta tables covers the Delta side of the same catalog.
The Iceberg REST catalog specification lists the endpoints these clients call.