上手 Apache Fluss：环境部署、建表读写全流程指南

一、什么是Fluss

Fluss（读作 “弗卢斯”）是 Apache 孵化器中的「面向实时分析的列式流存储」，由阿里 Flink 团队开源，全称 FLink Unified Streaming Storage，专为和 Flink 深度集成、构建秒级实时数仓 / 湖仓而设计。

1、简单定位

Kafka：消息队列，侧重 “数据流转发”，不支持更新、分析弱。 Paimon：流批一体湖格式，侧重 “历史 + 近实时存储”，分钟级延迟。 Fluss：毫秒级实时热层，侧重 “低延迟读写 + 实时更新 + 高效分析”，填补 Kafka/Paimon 之间的 “亚秒级分析型流存储” 空白。

2、核心能力（为什么用它）

• 列式流存储（Arrow 格式）

        按列存储，支持列裁剪、谓词下推，只读需要的列，比 Kafka 快 5–10 倍，大幅降低 IO / 网络成本。

• 毫秒级实时更新 & 点查

         支持 UPSERT/DELETE、部分列更新、CDC 订阅、维表点查，Kafka 做不到高效更新。

• 湖流一体（Fluss + Paimon）

        实时数据写 Fluss（热数据，亚秒级），自动分层到 Paimon（冷 / 历史数据，低成本）；统一 SQL 访问流 + 批。

• Flink 原生集成，解决状态爆炸

        提供 Delta Join：双流 Join 不用把全量数据存在 Flink State（避免几十 TB 状态），把状态下推到 Fluss 存储，大幅降本增效。

• AI 实时特征存储

        支持向量 / 多模态数据，低延迟提供实时特征，替代传统 Redis 特征库Apache Fluss (Incubating)。

3、典型架构（你当前场景）

数据接入 → Fluss（实时流存储，毫秒级） → Flink（计算） → Paimon（湖存储，历史） 你在 Flink SQL 里用 Fluss 表，就是把它当实时表 / 消息队列 / 维表来用，比 Kafka 更适合 SQL 分析与更新场景。

4、关键对比

二、安装与部署（Deploying Local Cluster | Apache Fluss™ (Incubating)https://fluss.apache.org/docs/install-deploy/deploying-local-cluster/）

要求​

Fluss 可在所有类 UNIX 环境中运行，例如 Linux、Mac OS X。 在开始设置系统之前，请确保你的测试机器上安装了以下软件：

• Java 11 或更高版（Archived OpenJDK GA Releases）（不支持 Java 8;强烈推荐使用 Java 17 或更高版本以获得更好的性能）

如果你的集群不满足这些软件需求，你需要安装或升级它。

JAVA_HOME配置​

Fluss 要求环境变量必须设置为 测试机器并指向你 Java 安装的目录。JAVA_HOME

Fluss装置​

去下载页面（apache-incubator-fluss安装包下载-开源镜像站-阿里云）下载最新的Fluss版本。一定要选Fluss 与你的 Java 版本匹配的包。下载最新版本后，解压：

tar -xzf fluss-0.9.1-incubating-bin.tgz
cd fluss-0.9.1-incubating/

启动Fluss​

您可以通过执行以下命令启动 Fluss 本地集群：

JAVA_HOME=/opt/installs/jdk-11.0.1 ./bin/local-cluster.sh start

之后，Fluss开始。

与Fluss的互动​

Fluss 本地集群启动后，你可以使用 Fluss 客户端（目前只支持 Flink SQL 客户端）与 Fluss 交互。 以下子章节将教你如何使用 Flink SQL 客户端与 Fluss 交互。

Flink SQL 客户端​

使用 Flink SQL 客户端与 Fluss 交互。

准备​

你可以启动一个独立的Flink集群，参见Flink环境准备https://fluss.apache.org/docs/engine-flink/getting-started/#preparation-when-using-flink-sql-client

注意：确保Fluss Flink连接器jarhttps://fluss.apache.org/docs/engine-flink/getting-started/#dependencies已经复制到你Flink主页的目录中。lib

使用 Flink SQL 客户端时的准备​

• 下载 Flink

Flink 可在所有类 UNIX 环境中运行，即 Linux、Mac OS X 和 Cygwin（Windows版）。 如果你还没下载 Flink，可以下载 Flink的二进制版本（apache-flink-flink-1.20.3安装包下载-开源镜像站-阿里云），然后用以下命令解压归档。

tar -zxvf flink-1.20.3-bin-scala_2.12.tgz -C /opt/installs

• 复制Fluss Flink捆绑罐

从上方的依赖部分下载Fluss Flink连接器JAR，并将其复制到你Flink主页的目录中。lib

cp fluss-flink-1.20-0.9.1-incubating.jar <FLINK_HOME>/lib/

注释

        如果你使用 Amazon S3、Aliyun OSS 或 HDFS（Hadoop 分布式文件系统）作为 Fluss 的远程存储， 你应该下载对应的Fluss文件系统JAR（见Filesystems），并复制到你Flink家的图书馆目录。

• 创建本地集群

要启动本地集群，运行 Flink 自带的 bash 脚本：

JAVA_HOME=/opt/installs/jdk-11.0.1 ./bin/start-cluster.sh

你应该能进入 localhost：8081 的网页界面，查看 Flink 仪表盘，确认集群已运行。你也可以用以下命令查看其状态：

ps aux | grep flink

• 启动一个SQL客户端

要快速启动SQL客户端，您可以使用提供的脚本：

<FLINK_HOME>/bin/sql-client.sh
或者
JAVA_HOME=/opt/installs/jdk-11.0.1 ./bin/sql-client.sh

• 停止集群

要快速停止集群和所有运行中的组件，您可以使用提供的脚本：

<FLINK_HOME>/bin/stop-cluster.sh

创建目录​

你可以使用以下SQL语句来创建目录。

Flink SQL

CREATE CATALOG fluss_catalog WITH (
'type' = 'fluss',
'bootstrap.servers' = 'localhost:9123'
);

该语句意思为：在
Flink SQL 中注册一个
Fluss 元数据目录（Catalog），让 Flink 能够连接 Fluss 服务，直接管理 Fluss 里面的所有流表、主键表，不用手动定义外部表。

逐段拆解

1.CREATE CATALOG fluss_catalog

• CREATE CATALOG：Flink SQL 语法，创建 / 注册一个外部元数据目录；
• fluss_catalog：自定义 Catalog 名称，后续用 USE CATALOG fluss_catalog; 切换使用。

2.WITH (…) 固定语法，存放当前 Catalog 的连接配置参数。

3.'type' = 'fluss' 标识这个 Catalog 的类型是 Fluss，Flink 会加载 Fluss 专用的 Catalog 连接器去建立连接。

4.'bootstrap.servers' = 'localhost:9123' Fluss 服务的访问地址：

• localhost：Fluss 部署的主机名 / IP；
• 9123：Fluss Coordinator 默认对外端口； 多节点集群写法示例：hadoop11:9123,hadoop12:9123

注释

创建表格​

Flink SQL

USE CATALOG fluss_catalog;

Flink SQL

CREATE TABLE pk_table (
shop_id BIGINT,
user_id BIGINT,
num_orders INT,
total_amount INT,
PRIMARY KEY (shop_id, user_id) NOT ENFORCED
) WITH (
'bucket.num' = '4'
);

数据写入​

要将新数据添加到表中，你可以使用批量模式或流式传输模式：INSERT INTO

Flink SQL

— Execute the flink job in batch mode for current session context
SET 'execution.runtime-mode' = 'batch';

Flink SQL

— use tableau result mode
SET 'sql-client.execution.result-mode' = 'tableau';

Flink SQL

INSERT INTO pk_table VALUES
(1234, 1234, 1, 1),
(12345, 12345, 2, 2),
(123456, 123456, 3, 3);

要在 Flink 流作业中用主键（1234， 1234）更新数据记录，请使用以下 UPDATE 语句：

Flink SQL

— should run in batch mode
UPDATE pk_table SET total_amount = 4 WHERE shop_id = 1234 and user_id = 1234;

要用主键删除数据记录，请使用：(12345, 12345)DELETE FROM

Flink SQL

— should run in batch mode
DELETE FROM pk_table WHERE shop_id = 12345 and user_id = 12345;

数据读取​

要检索使用主键的数据，可以通过对主键应用滤波器来进行点查询：(1234, 1234)

Flink SQL

— should run in batch mode
SELECT * FROM pk_table WHERE shop_id = 1234 and user_id = 1234;

要预览表中数据的子集，可以使用一个子句。LIMIT

Flink SQL

— should run in batch mode
SELECT * FROM pk_table LIMIT 10;

Fluss 支持在 flink 流作业中处理增量数据读取：

Flink SQL

— Submit the flink job in streaming mode for current session.
SET 'execution.runtime-mode' = 'streaming';

Flink SQL

— reading changelogs from the primary-key table from beginning.
SELECT * FROM pk_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

更多关于Fluss的作品​

目录创建后，你可以用 Flink SQL 客户端做更多 Fluss，比如创建表格、插入数据、查询数据等。 更多详情请参见Flink入门https://fluss.apache.org/docs/engine-flink/getting-started/指南