Connecting Query Engines

Any engine that speaks the Iceberg REST catalog protocol can read and write Arraylake Iceberg tables. Every engine needs the same three settings:

Setting	Value
Catalog URI	https://api.earthmover.io/iceberg
Warehouse	your organization name (e.g. my-org)
Token	an Arraylake bearer token

The warehouse is the org, so one catalog covers everything you can reach in it. Inside it, namespaces are flat and tables are addressed <namespace>.<table>. The examples below use the warehouse my-org and a namespace called my-tables. Nested namespaces are not supported — see Namespaces.

Use api-client tokens for engines

Both your personal login token and api-client tokens (ema_...) work as the catalog token. For query engines and long-running sessions, use an ema_... api-client token: engines hold the token statically for the life of the session, and user OAuth tokens expire after a few hours. The examples below use ema_... as a placeholder — substitute a real api-client token.

Arraylake vends temporary storage credentials to engines scoped to the table being loaded, so you never configure cloud credentials yourself. PyIceberg and DuckDB request vended credentials automatically; Spark needs one extra header configured (shown below). Vended credentials are valid for one hour and carry their expiry (s3.session-token-expires-at-ms); engines can refresh without reloading the table via GET .../tables/{table}/credentials, but not every engine does so automatically — if a job outlives its credentials, reload the table.

PyIceberg

Via the Arraylake client

The simplest path if you already use the arraylake Python client (installed with pip install 'arraylake[iceberg]'). get_iceberg returns a fully configured RestCatalog on your org's warehouse:

from arraylake import Client

client = Client()
catalog = client.get_iceberg("my-org")

Standalone

You can also configure PyIceberg directly, without the Arraylake client — useful in environments where you only want pyiceberg installed:

from pyiceberg.catalog.rest import RestCatalog

catalog = RestCatalog(
    name="my-org",
    uri="https://api.earthmover.io/iceberg",
    warehouse="my-org",
    token="ema_...",
    **{"header.X-Iceberg-Access-Delegation": "vended-credentials"},
)

Create, write, and read tables

Create the namespace once (from the client, the web app, or the catalog itself), then create tables in it:

import pyarrow as pa

data = pa.table(
    {
        "station": pa.array(["KSEA", "KPDX", "KSFO"], pa.string()),
        "temp_c": pa.array([11.5, 13.1, 14.2], pa.float64()),
    }
)

catalog.list_namespaces()  # [('my-tables',), ...] — the namespaces you can see in the org

table = catalog.create_table("my-tables.observations", schema=data.schema)
table.append(data)

# scan back to pandas
df = catalog.load_table("my-tables.observations").scan().to_pandas()

# filtered scan with predicate pushdown
hot = table.scan(row_filter="temp_c > 13").to_pandas()

A namespace created through the catalog lands on the org's default bucket unless the request names another one:

catalog.create_namespace("my-raw-tables", properties={"arraylake.bucket": "my-other-bucket"})

DuckDB

DuckDB (>= 1.3) can attach a warehouse as a database via the iceberg extension. Create an ICEBERG secret with your token, then ATTACH the org:

INSTALL iceberg;
LOAD iceberg;

CREATE SECRET al (
    TYPE ICEBERG,
    TOKEN 'ema_...',
    ENDPOINT 'https://api.earthmover.io/iceberg'
);

ATTACH 'my-org' AS warehouse (TYPE iceberg, SECRET al);

-- each namespace is a schema, holding that namespace's tables
SHOW ALL TABLES;

SELECT station, avg(temp_c) AS mean_temp
FROM warehouse."my-tables".observations
GROUP BY station
ORDER BY mean_temp DESC;

Every namespace you can see is attached at once, so you can join across namespaces without a second ATTACH. Iceberg's version history is queryable too:

SELECT * FROM iceberg_snapshots('warehouse."my-tables".observations');

Spark

Configure the Iceberg Spark runtime and point a REST catalog at Arraylake. Note the X-Iceberg-Access-Delegation header — unlike PyIceberg and DuckDB, Spark does not send it by default, and without it the catalog will not vend storage credentials:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("arraylake-iceberg")
    .config(
        "spark.jars.packages",
        "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.9.2,"
        "org.apache.iceberg:iceberg-aws-bundle:1.9.2",
    )
    .config("spark.sql.catalog.al", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.al.type", "rest")
    .config("spark.sql.catalog.al.uri", "https://api.earthmover.io/iceberg")
    .config("spark.sql.catalog.al.warehouse", "my-org")
    .config("spark.sql.catalog.al.token", "ema_...")
    .config("spark.sql.catalog.al.header.X-Iceberg-Access-Delegation", "vended-credentials")
    .config(
        "spark.sql.extensions",
        "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
    )
    .getOrCreate()
)

Then use ordinary Spark SQL against the al catalog, where each namespace is an Arraylake Iceberg namespace (backtick-quote names containing -):

spark.sql("SHOW NAMESPACES IN al").show()   # the namespaces you can see in my-org
spark.sql("""
    CREATE TABLE IF NOT EXISTS al.`my-tables`.observations (
        station STRING,
        temp_c DOUBLE
    ) USING iceberg
""")
spark.sql("INSERT INTO al.`my-tables`.observations VALUES ('KSEA', 11.5), ('KPDX', 13.1)")
spark.sql("SELECT * FROM al.`my-tables`.observations").show()

CREATE NAMESPACE al.my_new_tables also works — it creates the namespace on the org's default bucket — but only single-level names: CREATE NAMESPACE al.raw.obs fails. To place it on another bucket, set the property:

spark.sql("CREATE NAMESPACE al.my_new_tables WITH DBPROPERTIES ('arraylake.bucket' = 'my-other-bucket')")

For a namespace on GCS, replace iceberg-aws-bundle with org.apache.iceberg:iceberg-gcp-bundle:1.9.2.

polars

polars scans Iceberg tables lazily from a PyIceberg table handle, with predicate and projection pushdown:

import polars as pl

table = catalog.load_table("my-tables.observations")

df = (
    pl.scan_iceberg(table)
    .filter(pl.col("temp_c") > 13)
    .select("station", "temp_c")
    .collect()
)

pandas

Any PyIceberg scan converts straight to a pandas DataFrame:

df = catalog.load_table("my-tables.observations").scan().to_pandas()

Scans accept row filters and column selections so you only pull the data you need:

df = (
    catalog.load_table("my-tables.observations")
    .scan(row_filter="temp_c > 13", selected_fields=("station", "temp_c"))
    .to_pandas()
)

Other engines

Trino, Snowflake, ClickHouse, StarRocks, and many other systems support Iceberg REST catalogs with static bearer token authentication. Configure them with the same URI, warehouse, and token shown above. See the catalog reference for the exact set of supported REST endpoints and authentication details.

note

The catalog does not support the deprecated Iceberg oauth2 credential flow (POST /v1/oauth/tokens). Configure engines with a static bearer token, not a credential/client-secret pair.