Apache Flink writes to Apache Iceberg tables through a catalog, and Apache Polaris is an Iceberg REST Catalog implementation, so Flink connects to it with 'catalog-type'='rest' and an OAuth client credential. Polaris authenticates the job, enforces what it may write, and hands back short-lived storage credentials. Every successful Flink checkpoint becomes one Iceberg snapshot, which is the single most important thing to understand before putting this in production.
Streaming ingestion into a lakehouse is mostly a solved problem until somebody asks the second question. The first question is how to get Kafka records into Iceberg. The second is how the rest of the company reads those tables five minutes later without going through Flink.
That second question is what the catalog answers. Flink writes files and commits them, Polaris records what the table now consists of and who may touch it, and Spark, Trino, StarRocks or Dremio read the result without knowing a Flink job exists.
This walks through the connection, the configuration that matters, what happens at commit time, the permission model, and the failure modes that only appear after a week of running.
Four exchanges, and the order matters more than it looks.
The job authenticates to Polaris with a client ID and secret and receives a bearer token. It asks for a table, and Polaris returns the table metadata location plus, if you asked for it, temporary storage credentials scoped to that table's path. The TaskManagers then write Parquet files straight to object storage using those credentials. At checkpoint, the job commits the new files back to Polaris.
Data never passes through the catalog. That is worth stating plainly because people size Polaris as though it were in the ingestion path. It is not. It handles metadata operations, roughly one round trip per table per checkpoint, and a streaming job checkpointing every five minutes generates less catalog traffic in a day than a busy BI dashboard does in an hour.
The Iceberg Flink runtime jar, matched to your Flink minor version, and nothing else for the catalog itself:
# Flink 1.20 example. Match the artifact to your Flink minor version.
wget https://repo1.maven.org/maven2/org/apache/iceberg/\
iceberg-flink-runtime-1.20/<iceberg-version>/\
iceberg-flink-runtime-1.20-<iceberg-version>.jar
# plus the filesystem bundle for your object store, for example
flink-s3-fs-hadoop-<flink-version>.jarThe version pairing is the thing that costs people an afternoon. The Iceberg Flink runtime is published per Flink minor version, and a jar built for 1.19 against a 1.20 cluster fails with a class loading error that says nothing useful about versions.
You do not need a Polaris client library. Polaris speaks the Iceberg REST Catalog specification, so the standard REST catalog implementation in the Iceberg runtime is the client.
This is the whole integration:
CREATE CATALOG polaris WITH (
'type' = 'iceberg',
'catalog-type' = 'rest',
'uri' = 'https://polaris.internal:8181/api/catalog',
'warehouse' = 'analytics_catalog',
'credential' = '<client-id>:<client-secret>',
'scope' = 'PRINCIPAL_ROLE:ALL',
'header.X-Iceberg-Access-Delegation' = 'vended-credentials',
'token-refresh-enabled' = 'true'
);
USE CATALOG polaris;Four of those properties deserve explanation.
Polaris exposes two APIs. The management API, under /api/management, creates catalogs, principals and grants. The Iceberg REST API, under /api/catalog, is what engines speak. Pointing Flink at the base host or at the management path produces a 404 that is easy to misread as a networking problem.
In a Hadoop or Hive setup, warehouse is a filesystem location. In a REST catalog it identifies which catalog on the server you want. Set it to the name of the Polaris catalog. Storage locations are the catalog's business, not the client's, and that indirection is much of the point.
Format is client-id:client-secret, and the Iceberg client exchanges it for a bearer token at the token endpoint under the catalog URI. With token-refresh-enabled set to true the client refreshes before expiry, which is what keeps a job running for weeks from dying on a stale token.
Never put these literals in a SQL file that lives in git. Flink SQL supports environment substitution through the job launcher, and the sensible pattern is to inject them from whatever secret store you already run.
header.X-Iceberg-Access-Delegation: vended-credentials tells Polaris the client can accept temporary storage credentials. Without it the TaskManagers need their own standing access to the bucket, which defeats the reason to centralise authorisation in a catalog at all.
With it, Polaris returns credentials scoped to the table's path and lifetime, and revoking access in Polaris is enough to stop the job writing. Without it, revoking in Polaris changes nothing about what the TaskManagers can do with their own keys.
Worth walking through once, because every auth failure you will see is one of these steps.
POST /api/catalog/v1/oauth/tokens
grant_type=client_credentials
client_id=...&client_secret=...&scope=PRINCIPAL_ROLE:ALL
-> { access_token, expires_in }
GET /api/catalog/v1/analytics_catalog/namespaces/events/tables/clicks
Authorization: Bearer <token>
X-Iceberg-Access-Delegation: vended-credentials
-> metadata-location, config, storage-credentialsA 401 means the client credential is wrong or the token expired without refresh. A 403 means the principal authenticated fine and lacks the privilege, which is a grant problem rather than a credential problem. Those two are worth separating in your head before you start debugging, because they lead to completely different places.
CREATE DATABASE IF NOT EXISTS polaris.events;
CREATE TABLE polaris.events.clicks (
event_id STRING,
user_id BIGINT,
event_time TIMESTAMP(6),
url STRING,
PRIMARY KEY (event_id) NOT ENFORCED
) PARTITIONED BY (`url`)
WITH (
'format-version' = '2',
'write.upsert.enabled' = 'false'
);A note on partitioning for streaming tables. Partitioning by a high cardinality column produces a file per partition per checkpoint, which at a five minute checkpoint interval is a small file problem by Thursday. Partition by something coarse, usually a day or hour bucket on the event time, and let compaction handle the rest.
Iceberg's hidden partitioning helps here. You partition by days(event_time) and queries filtering on event_time prune correctly without anyone writing a partition predicate. That is a real difference from the Hive-era tables most streaming pipelines were built against.
This is the part that separates a Flink and Iceberg pipeline that works from one that quietly becomes unmanageable.
The Iceberg sink buffers records, flushes and uploads data files at checkpoint, and commits them to the catalog when the checkpoint completes. There is no separate commit interval. The checkpoint interval is the commit interval.
SET 'execution.checkpointing.interval' = '5min';
SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
INSERT INTO polaris.events.clicks
SELECT event_id, user_id, event_time, url FROM kafka_clicks;Exactly-once is the default for the Iceberg sink and it comes from this design. The commit carries the checkpoint ID in the snapshot summary, so a replay after failure can recognise a commit that already landed rather than duplicating it.
Set the interval by asking what freshness the readers actually need. Ten seconds sounds responsive and produces 8,640 snapshots a day per table. Five minutes produces 288, which nightly expiry handles without effort. The gap between those two numbers is the whole argument.
The sink exposes elapsedSecondsSinceLastSuccessfulCommit, and it is the one to alert on. A Flink checkpoint can succeed while the Iceberg commit fails, which leaves you with a job that looks healthy and a table that has stopped moving.
If your checkpoint interval is five minutes, alert when that gauge exceeds something like an hour. It catches the failure mode that job status alone cannot.
A streaming table accumulates snapshots at a rate no batch table ever does, and every one carries a manifest list. Left alone, planning slows down and metadata grows without bound.
Schedule expiry against streaming tables far more aggressively than against batch tables:
ALTER TABLE polaris.events.clicks SET TBLPROPERTIES (
'history.expire.max-snapshot-age-ms' = '21600000', -- 6 hours
'history.expire.min-snapshots-to-keep' = '20',
'write.metadata.delete-after-commit.enabled' = 'true',
'write.metadata.previous-versions-max' = '50'
);The last two matter as much as the first two on a streaming table. Every commit writes a new metadata JSON file, and at 288 commits a day that is 105,000 files a year if nothing removes them. Enabling delete-after-commit keeps the tracked set bounded.
The mechanics of what expiry removes, and what it refuses to remove, are covered in the snapshot expiration article. The short version for streaming: run it hourly, keep a floor of at least a few dozen snapshots, and never set the window shorter than your longest running reader.
Change data capture into Iceberg needs updates, not appends. Flink handles this with upsert mode:
INSERT INTO polaris.events.user_state
/*+ OPTIONS('upsert-enabled'='true') */
SELECT user_id, status, updated_at FROM cdc_stream;Requirements are specific. The table must be format version 2, and it must declare a primary key or identifier fields, because that is what defines row identity.
Understand the cost before you enable it. Upsert mode writes equality delete files alongside data files, and every reader has to apply those deletes at scan time. Accumulate enough of them and read performance falls off noticeably. Compaction rewrites them away, so an upsert pipeline without scheduled compaction is a pipeline with a read performance cliff in its future.
If your stream is append-only, leave upsert off. It is not a free improvement.
The reverse direction works too, which surprises people who think of Iceberg as a batch destination:
SELECT * FROM polaris.events.clicks
/*+ OPTIONS('streaming'='true', 'monitor-interval'='10s') */;
-- start from a known point instead of the whole table
SELECT * FROM polaris.events.clicks
/*+ OPTIONS('streaming'='true', 'monitor-interval'='10s',
'start-snapshot-id'='3821550127947089987') */;Flink polls the table for new snapshots and emits the records they added. The latency floor is the producing job's checkpoint interval, so an Iceberg table is not a replacement for Kafka when you need sub-second delivery. It is a good fit for a pipeline stage where minutes are acceptable and you want the intermediate state to be queryable by everything else.
Iceberg ships a TableMaintenance API for Flink that runs maintenance natively rather than requiring a separate Spark cluster:
TableMaintenance.forTable(env, tableLoader, lockFactory)
.add(ExpireSnapshots.builder()
.scheduleOnCommitCount(10)
.maxSnapshotAge(Duration.ofHours(6)))
.add(RewriteDataFiles.builder()
.scheduleOnDataFileCount(100))
.append();The TriggerLockFactory is not optional. It serialises maintenance tasks so two of them cannot run against the same table at once, and Iceberg ships JDBC and ZooKeeper implementations. The documentation is explicit that the lock is needed even for a single job, because multiple instances of the same task can otherwise conflict.
This is genuinely useful for teams whose only batch engine exists to run Iceberg maintenance. Removing that Spark cluster is a real operational simplification.
Polaris grants run through a chain: a principal is assigned a principal role, the principal role is granted a catalog role, and the catalog role holds privileges on catalogs, namespaces and tables. A streaming job needs a principal at the bottom of that chain and very little at the top.
# create the identity the Flink job uses
polaris principals create flink_ingest
polaris principal-roles create ingest_writer
polaris principal-roles grant ingest_writer --principal flink_ingest
# a catalog role scoped to one namespace
polaris catalog-roles create --catalog analytics_catalog events_writer
polaris privileges namespace grant \
--catalog analytics_catalog --catalog-role events_writer \
--namespace events TABLE_WRITE_DATA
polaris catalog-roles grant events_writer \
--catalog analytics_catalog --principal-role ingest_writerTwo habits worth adopting. Give each job its own principal rather than sharing one across the pipeline, because a shared principal makes an audit log useless and makes revocation a coordination exercise. And do not grant TABLE_CREATE to a production streaming job. Create tables through a deliberate process so schema changes are reviewed, not applied by a restarted job at three in the morning.
The full grant model, including how credential vending derives its scope from these same privileges, is covered in the Polaris access control article.
Flink supports 'catalog-type'='hadoop', which needs no service at all. It writes metadata files into a directory and treats the newest one as the table. For a laptop experiment it is the fastest thing to set up, and for anything else it is a trap worth naming.
A Hadoop catalog resolves the current table version by listing a directory and picking the highest numbered metadata file. Object storage does not give you an atomic rename, so two writers committing at the same moment can both believe they won. The Iceberg project has said for years that the Hadoop catalog is not safe for concurrent writes on S3, and a streaming job plus a compaction job is concurrent writers.
Beyond correctness, three things follow from having a catalog service. Authorisation gets a place to live, so a grant means something. Credential vending becomes possible, so engines stop holding standing keys. And other engines can find your tables by name rather than by someone sharing a storage path in a document.
The catalog is also the piece you will change least often, so choosing one on the first day of a streaming project is cheaper than migrating a hundred jobs later.
The Polaris repository ships a Flink guide with a working compose file, which is the fastest way to see the whole loop without provisioning anything:
# from a clone of the Polaris repository
export S3_ENDPOINT=http://rustfs:9000
docker compose \
-f site/content/guides/rustfs/docker-compose.yml \
-f site/content/guides/flink/docker-compose.yml up --build
# then open the Flink SQL client inside the jobmanager
docker exec -it $(docker ps -q --filter name=jobmanager) ./bin/sql-client.shThat brings up Polaris, a Flink cluster and an S3-compatible store on your machine. Register the catalog, create a table, insert a couple of rows, and select them back.
The official guide writes the catalog with 'catalog-impl' = 'org.apache.iceberg.rest.RESTCatalog' rather than 'catalog-type' = 'rest'. Both reach the same implementation. The explicit class name is useful when you also need to pin io-impl and S3 endpoint settings for a non-AWS store, which the local setup does.
One detail from that guide is worth carrying into production thinking. It tells you to allow about ten seconds between the insert and the select, because Iceberg only makes data visible after a checkpoint commits. That is the same rule as everything in the previous section, just small enough to watch happen.
Then query the same table from Spark or PyIceberg against the same Polaris instance. Two engines, two processes, one catalog, and neither knows the other exists. Reproducing that locally is what makes the architecture stop being abstract.
Somebody sets thirty seconds during development because it makes testing faster, and it ships. Six months later the table has a million snapshots and planning takes minutes. Pick the interval from what the readers need, and write down why.
Equality delete files accumulate, readers slow down gradually, and the diagnosis is difficult because nothing failed. Schedule compaction from the start on any table taking upserts.
The job works, which is why nobody notices. But the TaskManagers hold standing bucket access, and the permission model you built in Polaris governs metadata only. Check for the delegation header before you assume Polaris is enforcing anything about data.
A consequence of granting TABLE_CREATE. A job restarts against a table that was dropped, recreates it with whatever schema the current job version implies, and the downstream consumers find out later. Take the privilege away.
The job is running, checkpoints are succeeding, commits have not landed in six hours. Only elapsedSecondsSinceLastSuccessfulCommit catches this.
Dremio co-created Apache Polaris and donated it to the Apache Software Foundation, and Dremio's Open Catalog is a Polaris implementation. In practice that means a table your Flink job writes through Polaris is queryable from Dremio with no registration step, no copy, and no separate metadata definition.
The general shape is worth stating even if you use neither. Ingestion belongs to a streaming engine, serving belongs to a query engine, and the catalog is what lets them disagree about everything else while agreeing about what the table is.
If you take one decision away from this, make it the checkpoint interval, and make it before the pipeline ships rather than after.
Everything else here is adjustable later. Grants can be tightened, compaction can be added, credentials can be rotated. A table with two million snapshots because somebody picked ten seconds in a development environment is a genuinely tedious thing to repair, and the repair starts with changing the number you should have picked on day one.
No. Polaris implements the Iceberg REST Catalog specification, so the standard REST catalog support in the Iceberg Flink runtime is the client. You set 'catalog-type'='rest' and point it at the Polaris catalog URI.
Once per successful checkpoint, per sink. There is no independent commit interval. If you want fewer snapshots, lengthen the checkpoint interval, accepting the matching increase in end-to-end latency.
Yes, with checkpointing enabled. The Iceberg sink records the checkpoint ID in the snapshot summary, so a commit that already landed is recognised on replay rather than repeated.
Yes, with 'streaming'='true' and a monitor-interval. Flink polls for new snapshots and emits the records they added. The latency floor is the producing job's checkpoint interval, so this is a minutes-scale mechanism rather than a Kafka replacement.
The management API, under /api/management, is where you create catalogs, principals, roles and grants. The Iceberg REST API, under /api/catalog, is what engines speak. Flink only ever talks to the second one, and pointing it at the wrong path is a common first-run error.
You need them if you want Polaris to control data access rather than just metadata access. Without the X-Iceberg-Access-Delegation header, the TaskManagers use their own standing storage credentials, and revoking a grant in Polaris will not stop them writing.
In development, sure. In production, no. Grant TABLE_WRITE_DATA and withhold TABLE_CREATE so schema changes go through a deliberate process rather than happening implicitly when a job restarts.
For a full treatment of the catalog layer, download Apache Polaris: The Definitive Guide by Alex Merced, Andrew Madson, and Tomer Shiran, free from Dremio.
To see these ideas applied in practice, explore Dremio’s Open Catalog, built on Apache Polaris.
Deploy agentic analytics directly on Apache Iceberg data with no pipelines and no added overhead.
By unifying data from diverse sources, simplifying data operations, and providing powerful tools for data management, Dremio stands out as a comprehensive solution for modern data needs. Whether you are a data engineer, business analyst, or data scientist, harnessing the combined power of Dremio and Apache Iceberg will undoubtedly be a valuable asset in your data management toolkit.
Editor’s note, September 2026. This post was published in September 2023 and some product details have changed since. Nessie is still available and self-deployable under the Apache-2.0 licence, and it remains the clearest implementation of catalog-level branching. It is not an Apache Software Foundation project, and its development has slowed considerably. For how the current […]
This blog helps you learn about table-driven access policies in Dremio Cloud and Dremio Software v24.1+.
