Connect Flink to Iceberg REST
Apache Gravitino exposes an Iceberg REST catalog (IRC) endpoint that any Iceberg-compatible engine can connect to directly, without installing a Gravitino-specific connector plugin. Flink connects to it with the standard Iceberg Flink runtime.
Gravitino enforces its full access-control model on every IRC request, and authorizes each request as the identity in the catalog definition. Whether that identity is one person or an account shared by many depends on whose credentials the catalog holds; see Authentication.
Quick start
The example signs you in with your own token from the identity provider, connects to the Gravitino
catalog sales_lake
as the Flink catalog sales, and reads storage through vended credentials. Put the jars listed in
Prerequisites in $FLINK_HOME/lib first, then start the SQL client with
$FLINK_HOME/bin/sql-client.sh.
CREATE CATALOG sales WITH (
'type' = 'iceberg',
'catalog-type' = 'rest',
'uri' = 'https://{iceberg_host}/iceberg',
'warehouse' = 'sales_lake',
'token' = '{token}',
'header.X-Iceberg-Access-Delegation' = 'vended-credentials'
);
USE CATALOG sales;
SHOW DATABASES;
SELECT * FROM q4.leads LIMIT 10;
The catalog must vend credentials for the read to reach storage; see Credential vending.
Prerequisites
- Apache Gravitino running with the Iceberg REST service enabled. See Iceberg REST catalog service for setup instructions.
- The Iceberg REST service published and reachable from your Flink environment. See Client access.
- The following JAR files on your Flink classpath:
iceberg-flink-runtime-1.20-1.11.0.jar(oriceberg-flink-runtime-1.19-1.10.2.jarfor Flink 1.19, the last Iceberg release built for it)iceberg-aws-bundle-1.11.0.jarflink-shaded-hadoop-2-uber.jar
The examples use Flink 1.20 with Iceberg 1.11.0, the Iceberg version the server runs. An older client still creates, reads and writes tables through the IRC, but Basic authentication needs Iceberg 1.8 or later, and server-side scan planning, where the IRC plans a scan and returns its file tasks, needs Iceberg 1.11 or later.
Flink takes every catalog setting, including storage access, in the CREATE CATALOG statement
rather than in a configuration file.
Configuration
Flink uses a cluster configuration file (flink-conf.yaml) for general settings. The Iceberg
catalog is registered per-session using a CREATE CATALOG SQL statement.
flink-conf.yaml
Set batch execution mode and result display.
execution.runtime-mode: batch
sql-client.execution.result-mode: tableau
tableau mode prints query results inline in the terminal. Without it, the Flink SQL client opens
results in a full-screen pager.
Then start the Flink SQL client:
$FLINK_HOME/bin/sql-client.sh
Register the catalog
At the Flink SQL client prompt, run a CREATE CATALOG statement. The one below connects with your
own token and vended credentials; the alternatives follow it.
CREATE CATALOG gravitino_irc WITH (
'type' = 'iceberg',
'catalog-type' = 'rest',
'uri' = 'https://{iceberg_host}/iceberg',
'warehouse' = '{catalog}',
'token' = '{token}',
'header.X-Iceberg-Access-Delegation' = 'vended-credentials'
);
warehouse is the name of the Gravitino catalog to connect to. The client sends it to the service,
receives that catalog's prefix, and sends every later request to that catalog. Without it, the
service serves its default catalog, if one is set; see
Select a catalog.
The catalog lasts for the SQL client session, so run CREATE CATALOG again in each new session.
Authentication
Every request to the IRC needs an identity; a request with none is rejected with 401. See
How to authenticate for the Gravitino side.
Use OAuth. Each Flink catalog signs in with exactly one identity, set in its CREATE CATALOG
statement, and sends it on every request: either a shared identity, from a client ID and secret or
a pre-issued token, or a person's own token. A catalog with a person's token needs no client ID or
secret. Basic authentication signs in as one Gravitino local account.
Where the catalog is defined decides who shares its identity. A catalog a person creates in their own SQL client session is theirs alone. A catalog defined for a shared Flink cluster or SQL gateway, or built into a job, identifies everyone who uses it as the same account.
Shared identity for Flink
Use a shared identity for jobs and pipelines that run as a service account. Every catalog that signs in with it is authorized as the same account, so grant that account in Gravitino only what those jobs need.
Client ID and client secret
Flink requests a token from the identity provider's token endpoint and requests a new one when it
expires. Replace the token line with:
'credential' = '{client_id}:{client_secret}',
'oauth2-server-uri' = '{token_endpoint_uri}',
'scope' = '{scope}',
oauth2-server-uri is the token endpoint, not the issuer.
Pre-issued token
A token issued to a service account, set in the token property the same way as a
personal token. Nothing renews it, so requests fail once it expires. Use it only
when Flink cannot reach the identity provider.
Token exchange settings
Add this to the client credential properties:
'token-exchange-enabled' = 'false',
The setting controls how Flink renews the client's token. With the default, true, Flink renews it
by exchanging the old token at the identity provider; an identity provider that does not support
token exchange, such as a default Keycloak realm, refuses with unsupported_grant_type and Flink
retries repeatedly. With false, Flink signs in again with the client ID and secret.
Per-user identity
A person gets a token from the identity provider, outside Flink, and sets it in the catalog's
token property, as in the Quick start. Flink sends it as the bearer token on every
request, and Gravitino authorizes each request as that person.
The token must carry the audience configured for Gravitino and a principal claim naming a member of the metalake. Nothing renews it, so requests fail once it expires; create the catalog again with a new token.
Basic authentication
A local user signs in with Basic credentials. Replace the
token line with:
'rest.auth.type' = 'basic',
'rest.auth.basic.username' = '{username}',
'rest.auth.basic.password' = '{password}',
Storage access
With header.X-Iceberg-Access-Delegation, as above, the IRC returns short-lived storage credentials
scoped to each table, together with the storage endpoint and region. The catalog must have a
credential provider; see Credential vending.
To use storage keys held by Flink instead, remove that line and add the keys:
's3.access-key-id' = '{access_key}',
's3.secret-access-key' = '{secret_key}',
For S3-compatible storage such as MinIO, also add the endpoint and path-style access when the catalog does not already send them:
's3.endpoint' = 'https://{minio_host}:9000',
's3.path-style-access' = 'true',
Examples
Use the catalog
USE CATALOG gravitino_irc;
List databases
SHOW DATABASES;
List tables
SHOW TABLES IN {namespace};
Query a table
SELECT * FROM {namespace}.{table};
Create a table
CREATE TABLE gravitino_irc.{namespace}.new_table (
id INT,
name STRING,
created_at TIMESTAMP
);
Insert data
INSERT INTO gravitino_irc.{namespace}.new_table VALUES (1, 'example', CURRENT_TIMESTAMP);
Gravitino connector vs. the IRC
| Feature | Gravitino engine connector | Iceberg REST |
|---|---|---|
| Engine plugin required | Yes | No |
| Gravitino access control | Yes | Yes |
| Supported engines | Trino, Spark, Flink, Daft | Any Iceberg-compatible engine |
| Credential vending | Varies | Yes (S3, GCS, OSS, ADLS) |