Ververica DocsVerverica Docs
Docs Home

Reading and Writing Apache Fluss™

6 min read

Configure VERA SQL jobs and Java applications to connect to Apache Fluss, write and read its tables, and read tiered tables with each catalog's credentials.

On this page

Applies to: the Fluss connector and the Fluss Java client, with client setups from Docker image registry.ververica.cloud/platform-images/fluss:1.0.0-vv-3

Overview

This manual covers configuring a VERA SQL job or a Java application to connect to Fluss, write to it, read from it, and read the tables Fluss tiers into a lake. Jobs run on Ververica Platform 3. Table kinds, their DDL, startup modes, and other table features are in the Apache Fluss documentation.

Prerequisites

  • A running Fluss cluster and its bootstrap address, <BOOTSTRAP_ADDRESS>. See Connecting to Fluss.
  • With SASL on the client listener, a client user, <CLIENT_USERNAME>, and its password. For the server's SASL configuration and its users, see Securing Fluss.
  • Pull access to registry.ververica.cloud, to run the setup installer from the Fluss image. See Obtaining Registry Access.

The Apache Fluss Catalog

A VERA SQL job reaches Fluss through a Fluss catalog. The SQL in this manual runs in a SQL draft on Ververica Platform 3: see Creating a Draft and Running Statements in the SQL Editor.

Connecting

Without authentication, the catalog needs the bootstrap address only:

SQL
1CREATE CATALOG fluss WITH (
2    'type'              = 'fluss',
3    'bootstrap.servers' = '<BOOTSTRAP_ADDRESS>'
4);

With SASL on the client listener, the catalog carries the client user's credentials. The password is a secret value, fluss-password here (see Creating Secret Values):

SQL
1CREATE CATALOG fluss WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<BOOTSTRAP_ADDRESS>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<CLIENT_USERNAME>',
7    'client.security.sasl.password'  = '${secret_values.fluss-password}'
8);
9
10USE CATALOG fluss;
11USE `<DATABASE>`;

Beyond the bootstrap address and its default database, the catalog accepts only client.security.* options and the lake credentials of Lake Credentials, and rejects any other option: the object store's options are set per query instead, as Options by Object Store shows. For the connector's options, see Ververica's Fluss connector page and the upstream Fluss connector options.

Lake Credentials

A read of a tiered table needs the lake catalog's credentials, which the Fluss server does not hand to clients. Add them to the catalog as the Iceberg catalog's own properties, each prefixed with iceberg. and each a secret value. For each tested lake catalog:

A Lakekeeper catalog, on any store:

SQL
1CREATE CATALOG `<LAKE_CATALOG_NAME>` WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<BOOTSTRAP_ADDRESS>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<CLIENT_USERNAME>',
7    'client.security.sasl.password'  = '${secret_values.fluss-password}',
8    'iceberg.token'                  = '${secret_values.iceberg-token}'
9);

A Polaris catalog, on Amazon S3:

SQL
1CREATE CATALOG `<LAKE_CATALOG_NAME>` WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<BOOTSTRAP_ADDRESS>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<CLIENT_USERNAME>',
7    'client.security.sasl.password'  = '${secret_values.fluss-password}',
8    'iceberg.credential'             = '${secret_values.iceberg-credential}',
9    'iceberg.s3.access-key-id'       = '${secret_values.s3-access-key-id}',
10    'iceberg.s3.secret-access-key'   = '${secret_values.s3-secret-access-key}'
11);

A JDBC catalog, on Amazon S3 or an S3-compatible store:

SQL
1CREATE CATALOG `<LAKE_CATALOG_NAME>` WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<BOOTSTRAP_ADDRESS>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<CLIENT_USERNAME>',
7    'client.security.sasl.password'  = '${secret_values.fluss-password}',
8    'iceberg.jdbc.password'          = '${secret_values.jdbc-password}',
9    'iceberg.s3.access-key-id'       = '${secret_values.s3-access-key-id}',
10    'iceberg.s3.secret-access-key'   = '${secret_values.s3-secret-access-key}'
11);

On an S3-compatible store, a read through a JDBC catalog also sets these JVM options in the job's Flink configuration:

YAML
1env.java.opts.all: "-Daws.requestChecksumCalculation=when_required -Daws.responseChecksumValidation=when_required"

A JDBC catalog, on Azure Blob Storage:

SQL
1CREATE CATALOG `<LAKE_CATALOG_NAME>` WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<BOOTSTRAP_ADDRESS>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<CLIENT_USERNAME>',
7    'client.security.sasl.password'  = '${secret_values.fluss-password}',
8    'iceberg.jdbc.password'          = '${secret_values.jdbc-password}',
9    'iceberg.adls.auth.shared-key.account.name' = '<STORAGE_ACCOUNT>',
10    'iceberg.adls.auth.shared-key.account.key'  = '${secret_values.adls-account-key}'
11);

With a JDBC lake catalog, name the Fluss catalog after the server's datalake.iceberg.name, the name the JDBC catalog files its tables under. For the server's lake catalog, see Configuring Lake Tiering.

Configuring Writes

A writer installs the flink-connector setup (see Installing Fluss Setups):

BASH
1docker run --rm -v "$PWD/<LOCAL_DIR>":/output \
2  registry.ververica.cloud/platform-images/fluss:1.0.0-vv-3 \
3  /opt/fluss/bin/setup/install.sh client flink-connector --output-dir /output/flink-connector

It uploads the JARs, as Uploading the JARs as Artifacts shows. A writer sends its rows to the Fluss server, which handles all object-storage I/O, so a writer needs no store credentials, whatever the cluster's object store.

A streaming SQL deployment then writes through the catalog:

SQL
1INSERT INTO <TABLE> SELECT * FROM <SOURCE_TABLE>;

Configuring Reads

A VERA SQL read sets its setups and, on some object stores, its store options. Install each setup as Installing Fluss Setups shows.

Dependencies

A reader fetches data from the cluster's object store itself, so it needs the filesystem setup of that store: fs-s3 for Amazon S3 or an S3-compatible store, fs-azure for Azure Blob Storage. Install it, then upload its JARs (see Uploading the JARs as Artifacts):

BASH
1docker run --rm -v "$PWD/<LOCAL_DIR>":/output \
2  registry.ververica.cloud/platform-images/fluss:1.0.0-vv-3 \
3  /opt/fluss/bin/setup/install.sh client fs-s3 --output-dir /output/fs-s3

A reader of tiered tables installs the lake-iceberg setup instead, and uploads all its JARs. Its --fs option names the object store: s3 on Amazon S3 or an S3-compatible store, abs on Azure Blob Storage. Its --catalog option names the lake catalog: rest for Lakekeeper and Polaris, jdbc-postgres for a JDBC catalog. For example, on Amazon S3 with Lakekeeper:

BASH
1docker run --rm -v "$PWD/<LOCAL_DIR>":/output \
2  registry.ververica.cloud/platform-images/fluss:1.0.0-vv-3 \
3  /opt/fluss/bin/setup/install.sh client lake-iceberg --fs s3 --catalog rest --output-dir /output/lake-iceberg
BASH
1docker run --rm -v "$PWD/<LOCAL_DIR>":/output \
2  registry.ververica.cloud/platform-images/fluss:0.9.1-vv-1 \
3  /opt/fluss/bin/setup/install.sh client fs-s3 --output-dir /output/fs-s3

Every job uploads the Fluss Flink connector its setups install, even though the VERA engine ships its own Fluss connector: the one a setup installs comes with your Fluss release, so it might be more recent than the engine's, and the engine's built-in Fluss catalog rejects some of the lake credential options of Lake Credentials, such as iceberg.token.

For the union, lake-only, and snapshot queries on a tiered table, see the Quickstart: Fluss on Ververica Platform 3. Any Iceberg-compatible engine, such as Spark, Trino, or StarRocks, also reads a tiered table by attaching to the same Iceberg catalog and warehouse that Fluss tiers into.

Options by Object Store

Lake credentials go in the catalog (see Lake Credentials), but the object store's options are set at read time, as a query hint. What a read adds depends on the cluster's object store (see Configuring Remote Storage):

Amazon S3, and Azure Blob Storage with an OAuth service principal: nothing. The server hands the client short-lived credentials for the store.

An S3-compatible store without STS: the store's endpoint and access key, as client.fs.s3.* options in a query hint on every read. The catalog rejects them, and the read forwards a hint's client.* options to the client. Each credential is a secret value:

SQL
1SELECT * FROM <TABLE> /*+ OPTIONS('scan.startup.mode' = 'full',
2                                  'client.fs.s3.endpoint' = '<S3_ENDPOINT>',
3                                  'client.fs.s3.access.key' = '${secret_values.store-access-key}',
4                                  'client.fs.s3.secret.key' = '${secret_values.store-secret-key}',
5                                  'client.fs.s3.region' = '<S3_REGION>',
6                                  'client.fs.s3.path-style-access' = 'true') */;

A read of a tiered table on such a store carries the same hint.

Azure Blob Storage with an account key only, when the cluster has no OAuth service principal: the account key as client.fs.fs.azure.account.key, in a query hint. The client strips the client.fs. prefix before handing a key to its filesystem, so the filesystem receives fs.azure.account.key:

SQL
1SELECT * FROM <TABLE> /*+ OPTIONS('client.fs.fs.azure.account.key' = '${secret_values.adls-account-key}') */;

The Java Client

A Java application connects to Fluss through the Fluss Java client.

Connecting

The application passes a client configuration to ConnectionFactory. With SASL on the client listener, the configuration carries the same client.security.* keys as the catalog:

JAVA
1import org.apache.fluss.client.Connection;
2import org.apache.fluss.client.ConnectionFactory;
3import org.apache.fluss.config.Configuration;
4
5Configuration conf = new Configuration();
6conf.setString("bootstrap.servers", "<BOOTSTRAP_ADDRESS>");
7conf.setString("client.security.protocol", "sasl");
8conf.setString("client.security.sasl.mechanism", "PLAIN");
9conf.setString("client.security.sasl.username", "<CLIENT_USERNAME>");
10conf.setString("client.security.sasl.password", "<CLIENT_PASSWORD>");
11Connection conn = ConnectionFactory.createConnection(conf);

Without SASL, set bootstrap.servers only. Writing needs nothing more. For the client API, see the Apache Fluss documentation.

Reading

A reading application fetches data from the cluster's object store itself, so it puts that store's filesystem setup on its classpath, next to the application and its Fluss client: fs-s3 for Amazon S3 or an S3-compatible store, fs-azure for Azure Blob Storage. Install the setup with --output-dir, as in Dependencies, then:

BASH
1java -cp "<APP_DIR>/*:<LOCAL_DIR>/fs-s3/*" <MAIN_CLASS>

On an S3-compatible store without STS, the application sets the store's client.fs.s3.* keys on its configuration:

JAVA
1conf.setString("client.fs.s3.endpoint", "<S3_ENDPOINT>");
2conf.setString("client.fs.s3.access.key", "<ACCESS_KEY_ID>");
3conf.setString("client.fs.s3.secret.key", "<SECRET_ACCESS_KEY>");
4conf.setString("client.fs.s3.region", "<S3_REGION>");
5conf.setString("client.fs.s3.path-style-access", "true");

On an Azure Blob Storage cluster with an account key only, it sets client.fs.fs.azure.account.key instead.

Troubleshooting

Connection refused or a DNS resolution error

The client cannot reach the bootstrap address. Check that the client runs on a network that resolves the services of the Fluss namespace, that the coordinator pod is ready, and that no NetworkPolicy blocks the bootstrap port.

ClassNotFoundException for FlussCatalogFactory or filesystem classes

The job lacks a setup: FlussCatalogFactory comes with the Fluss connector, and the filesystem classes with the fs-s3 or fs-azure setup, needed once a read reaches remote data. Add the setup's JARs, as Dependencies shows.

Authentication fails

The cluster has SASL on its client listener, and the client does not send a listed user's credentials. Add the client.security.* options to the catalog, as Connecting shows, or the same keys to the Java client's configuration.

IllegalStateException: Dynamic session credentials for Fluss on Azure Blob Storage

The cluster authenticates to Azure Blob Storage with an account key only, so the server has no delegation tokens to hand out. Pass the account key as client.fs.fs.azure.account.key, as Options by Object Store shows.

Reads fail on an S3-compatible store without STS

A read on such a store carries the store's endpoint and access key: in a query hint for a VERA SQL read, as Options by Object Store shows, or as client.fs.s3.* keys for the Java client, as Reading shows.

Further Reading

Was this helpful?