Skip to main content
Version: Next

Deploying Streaming Lakehouse

This guide covers how to deploy a Fluss cluster with Streaming Lakehouse capabilities. For conceptual overview, see Lakehouse Overview.

Prerequisites​

  1. A running Fluss cluster (see Deploying Distributed Cluster)
  2. A running Flink cluster (for the Tiering Service)
  3. Access to a data lake storage system (S3, HDFS, OSS, etc.)

Cluster Configuration​

You can enable Lakehouse storage through:

  1. Static configuration: Configure in server.yaml before starting the cluster
  2. Dynamic configuration: Enable at runtime using the set_cluster_configs procedure

Method 1: Static Configuration​

Configure lakehouse settings in server.yaml on all Fluss servers (CoordinatorServer and TabletServer).

Fluss follows a simple convention for these keys:

  • datalake.enabled: true explicitly enables lakehouse capability for the cluster. If it is left unset, configuring datalake.format alone also enables it; whenever datalake.enabled is true, datalake.format must also be set.
  • datalake.format selects the lake format for the cluster (paimon, iceberg, or lance).
  • Format-specific options use the datalake.<format>.* prefix. Fluss strips this prefix and passes the remaining keys straight to the corresponding lake catalog/client — for example datalake.paimon.metastore becomes metastore and datalake.iceberg.type becomes type. Any option supported by the lake catalog can be set this way.
  • Only configure options for the format selected by datalake.format.
note

The Tiering Service is an independent Flink job and needs the same datalake.* options passed as job arguments (see Starting the Tiering Service).

server.yaml
datalake.enabled: true
datalake.format: paimon
datalake.paimon.metastore: filesystem
datalake.paimon.warehouse: /path/to/paimon/warehouse

For Hive catalog:

server.yaml
datalake.enabled: true
datalake.format: paimon
datalake.paimon.metastore: hive
datalake.paimon.uri: thrift://<hive-metastore-host>:<port>
datalake.paimon.warehouse: hdfs:///path/to/warehouse

Method 2: Dynamic Configuration​

Enable lakehouse settings at runtime using Flink SQL:

Flink SQL
USE fluss_catalog;

CALL sys.set_cluster_configs(
  config_pairs => 'datalake.format', 'paimon',
                  'datalake.paimon.metastore', 'filesystem',
                  'datalake.paimon.warehouse', '/path/to/warehouse'
);

See set_cluster_configs for more details.

Adding Required JARs​

Fluss Server JARs​

Add JARs to ${FLUSS_HOME}/plugins/<format>/ based on your configuration:

ScenarioRequired JAR
Paimon with S3paimon-s3-<version>.jar, matching your Paimon version
Paimon with OSSpaimon-oss-<version>.jar, matching your Paimon version
Paimon Hive catalogFlink SQL Hive connector JAR

Starting the Tiering Service​

The Tiering Service is a Flink job that continuously tiers data from Fluss to the data lake. For architecture details, see Tiering Service.

Prerequisites​

  1. Download fluss-flink-tiering-1.0-SNAPSHOT.jar

Add the following to ${FLINK_HOME}/lib:

If using S3, OSS, or HDFS as Fluss's remote storage, also add the corresponding Fluss filesystem JAR.

Start the Service​

${FLINK_HOME}/bin/flink run /path/to/fluss-flink-tiering-1.0-SNAPSHOT.jar \
    --fluss.bootstrap.servers localhost:9123 \
    --datalake.format paimon \
    --datalake.paimon.metastore filesystem \
    --datalake.paimon.warehouse /tmp/paimon
note
  • You must pass all datalake.* options that were set in server.yaml as command-line arguments
  • For S3/cloud storage, include additional flags like --datalake.paimon.s3.endpoint, --datalake.paimon.s3.access-key, etc.
  • The Tiering Service is stateless—you can run multiple instances for scalability
  • Use -D to pass Flink configurations (e.g., -Dparallelism.default=3)
  • For complete examples with S3 configuration, see the Lakehouse Quickstart

Enabling Lakehouse for Tables​

Create tables with lakehouse storage enabled:

Flink SQL
CREATE TABLE my_table (
    id BIGINT PRIMARY KEY NOT ENFORCED,
    name STRING
) WITH (
    'table.datalake.enabled' = 'true',
    'table.datalake.freshness' = '1min'
);

Verification​

  1. Check the Tiering Service job is running in Flink Web UI
  2. After the freshness interval, query the lake table:
Flink SQL
-- Lake-only query
SELECT * FROM my_table$lake;

-- Union Read (real-time + historical)
SELECT * FROM my_table;

See Union Read for more details.