Connect Flink to Iceberg REST

Introduction

Apache Gravitino exposes an Iceberg REST catalog endpoint that any Iceberg-compatible engine can connect to directly, without installing a Gravitino-specific connector plugin. This page describes how to configure Apache Flink to use Gravitino's Iceberg REST (IRC) endpoint.

This integration uses the standard Apache Iceberg REST catalog specification. Gravitino enforces its full access-control model on all IRC requests.

Prerequisites

Apache Gravitino running with the Iceberg REST service enabled. See Iceberg REST catalog service for setup instructions.
The Gravitino IRC endpoint is accessible from your Flink environment. The default port is 9001.
The following JAR files on your Flink classpath:
- iceberg-flink-runtime-1.18-1.7.1.jar (or iceberg-flink-runtime-1.19-1.7.1.jar for Flink 1.19)
- iceberg-aws-bundle-1.7.1.jar
- flink-shaded-hadoop-2-uber.jar

This page uses Flink 1.18 with Iceberg 1.7.1.

Unlike Spark and Trino, Flink requires S3 connection properties to be specified in the catalog definition itself rather than in a separate 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.

$FLINK_HOME/conf/flink-conf.yaml
execution.runtime-mode: batch
sql-client.execution.result-mode: tableau

tableau mode prints query results inline in the terminal. Without it, Flink SQL Client opens results in a full-screen pager.

Start the Flink SQL client

$FLINK_HOME/bin/sql-client.sh

Register the catalog

At the Flink SQL Client prompt, run the following CREATE CATALOG statement. Replace {gravitino_host} with your Gravitino server address and supply your S3 credentials.

No authentication

Create the catalog without authentication
CREATE CATALOG gravitino_irc WITH (
  'type'                 = 'iceberg',
  'catalog-type'         = 'rest',
  'uri'                  = 'http://{gravitino_host}:9001/iceberg',
  'io-impl'              = 'org.apache.iceberg.aws.s3.S3FileIO',
  's3.region'            = 'us-east-1',
  's3.access-key-id'     = '{access_key}',
  's3.secret-access-key' = '{secret_key}'
);

Basic authentication

Create the catalog with basic authentication
CREATE CATALOG gravitino_irc WITH (
  'type'                     = 'iceberg',
  'catalog-type'             = 'rest',
  'uri'                      = 'http://{gravitino_host}:9001/iceberg',
  'rest.auth.type'           = 'basic',
  'rest.auth.basic.username' = '{username}',
  'rest.auth.basic.password' = '{password}',
  'io-impl'                  = 'org.apache.iceberg.aws.s3.S3FileIO',
  's3.region'                = 'us-east-1',
  's3.access-key-id'         = '{access_key}',
  's3.secret-access-key'     = '{secret_key}'
);

OAuth authentication

Create the catalog with OAuth
CREATE CATALOG gravitino_irc WITH (
  'type'                 = 'iceberg',
  'catalog-type'         = 'rest',
  'uri'                  = 'http://{gravitino_host}:9001/iceberg',
  'rest.auth.type'       = 'oauth2',
  'rest.auth.oauth2.token' = '{token}',
  'io-impl'              = 'org.apache.iceberg.aws.s3.S3FileIO',
  's3.region'            = 'us-east-1',
  's3.access-key-id'     = '{access_key}',
  's3.secret-access-key' = '{secret_key}'
);

See How to authenticate for Gravitino authentication configuration options.

The catalog registration persists for the duration of the SQL Client session. You must re-run CREATE CATALOG each time you start a new session.

Local development.

For local development with MinIO, add these properties to the catalog definition.

Add the MinIO S3 properties
  's3.endpoint'          = 'http://{minio_host}:9000',
  's3.path-style-access' = 'true',

See gravitino-irc-quickstart for a complete local development environment using MinIO.

Examples

Use the catalog

Use the catalog
USE CATALOG gravitino_irc;

List databases

List databases
SHOW DATABASES;

List tables

List tables in a namespace
SHOW TABLES IN {namespace};

Query a table

Query a table
SELECT * FROM {namespace}.{table};

Create a table

Create a table
CREATE TABLE gravitino_irc.{namespace}.new_table (
  id INT,
  name STRING,
  created_at TIMESTAMP
);

Insert data

Insert a row
INSERT INTO gravitino_irc.{namespace}.new_table VALUES (1, 'example', CURRENT_TIMESTAMP);

Gravitino connector vs. Iceberg REST

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)

Introduction​

Prerequisites​

Configuration​

flink-conf.yaml​

Start the Flink SQL client​

Register the catalog​

No authentication​

Basic authentication​

OAuth authentication​

Examples​

Use the catalog​

List databases​

List tables​

Query a table​

Create a table​

Insert data​

Gravitino connector vs. Iceberg REST​

Related​