流存储Fluss

功能说明

Fluss 主键表支持全增量一体化消费：先读取表中的历史全量快照数据，全量数据读取完毕后自动无缝切换到增量流式消费，保证数据的完整性和一致性。

消费模式

通过 scan.startup.mode 参数控制消费起始位点：

SQL 示例

-- 全增量一体化消费（默认模式）
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;

-- 从指定时间戳开始消费，通过 Hint 指定参数
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'timestamp', 'scan.startup.timestamp' = '2025-01-01 00:00:00') */;

-- 从最新位点开始消费
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

日志订阅

Fluss 主键表天然支持 CDC（Change Data Capture）日志订阅，下游 Flink 作业可以实时消费到完整的变更事件流（INSERT、UPDATE_BEFORE、UPDATE_AFTER、DELETE），无需额外开启 Binlog 配置。

这意味着：

• 主键表的所有变更都会自动产生 CDC 日志。

• 下游可直接订阅 Fluss 主键表获取完整的变更流，用于实时数据同步或流式 ETL。

SQL 示例

-- 直接读取主键表即可获取 CDC 变更流
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;
-- 输出包含完整的 INSERT / UPDATE / DELETE 事件

虚拟表（Virtual Table）

除了直接订阅 CDC 流之外，Fluss 还提供了虚拟表机制，通过在基础表名后追加特定后缀即可访问带有元数据的变更日志，无需额外存储数据。虚拟表是系统自动生成的，不需要用户手动创建。

-- 读取 orders 表的 changelog 虚拟表
SELECT * FROM `fluss_catalog`.`fluss_db`.`orders$changelog`;

-- 通过 Hint 指定从最新位点开始消费
SELECT * FROM `fluss_catalog`.`fluss_db`.`orders$changelog`
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

-- 过滤出删除事件
SELECT * FROM `fluss_catalog`.`fluss_db`.`orders$changelog`
WHERE _change_type = 'delete';

-- 读取 users 表的 binlog 虚拟表
SELECT * FROM `fluss_catalog`.`fluss_db`.`users$binlog`;

-- 获取更新事件中字段变更前后的值
SELECT
  `before`.user_name AS old_name,
  `after`.user_name  AS new_name,
  _commit_timestamp  AS change_time
FROM `fluss_catalog`.`fluss_db`.`users$binlog`
WHERE _change_type = 'update';

功能说明

Fluss 支持部分列更新（Partial Update），允许多个数据流独立地向同一张主键宽表写入各自关心的字段，Fluss 引擎会自动按主键进行字段级别的合并。这种方式无需在 Flink 中维护复杂的多流 Join 状态，避免了大状态问题。

使用场景

典型场景为实时用户画像宽表构建：不同的数据源（如推荐系统、曝光日志、点击日志、购物车数据）分别更新用户画像表中各自对应的字段，最终形成一张完整的用户 360 度视图。

使用限制

• 目标表必须为主键表。

• 每个数据流的写入字段必须包含完整的主键字段。

• 计算拓扑中不包含多流 Join 节点，直接由 Fluss 在存储层完成 Merge。

SQL 示例

-- 1. 创建宽表
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,
    imp_cnt   INT,
    click_cnt INT,
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

-- 2. 多流分别写入各自字段
-- 数据流 1：写入推荐分数
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, rec_score)
SELECT user_id, item_id, rec_score 
FROM `fluss_catalog`.fluss_db.recommendations;

-- 数据流 2：写入曝光次数
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, imp_cnt)
SELECT user_id, item_id, imp_cnt 
FROM `fluss_catalog`.fluss_db.impressions;

-- 数据流 3：写入点击次数
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, click_cnt)
SELECT user_id, item_id, click_cnt 
FROM `fluss_catalog`.fluss_db.clicks;

上述三条 INSERT 语句可作为独立的 Flink 作业运行，Fluss 存储层会自动将相同主键的字段合并成完整的行。

功能说明

Delta Join 是 Fluss 提供的新一代双流 Join 范式，将数据更新行为下沉到 Fluss 存储层完成，利用主键表和二级索引实现高效查找，无需在 Flink 中维护 Join State。

核心优势：

• 无 Join State：省去冗余的状态存储，避免大状态导致的 GC 问题。

• 低资源消耗：实测 Flink 内存与 CPU 消耗下降超 86%。

• 更稳定高效：避免状态膨胀带来的性能瓶颈。

与传统 Join 对比

使用限制

• 左右表必须为 Fluss 的主键表（支持分区表）。

• 分桶键（Bucket Key）需为主键前缀。

• Join Key 必须包含分桶键；若为分区表，还需包含分区键。

SQL 示例

1. 创建表

CREATE TABLE orders (
  user_id BIGINT,
  order_date DATE,
  order_id BIGINT,
  amount DECIMAL(10, 2),
  PRIMARY KEY (user_id, order_id, order_date) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

CREATE TABLE order_enhance (
  user_id BIGINT,
  order_id BIGINT,
  risk_level TINYINT,
  is_fraud BOOLEAN,
  update_time TIMESTAMP(3),
  PRIMARY KEY (user_id, order_id) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

2. 执行 Delta Join

SELECT * FROM orders o
INNER JOIN order_enhance e
ON o.user_id = e.user_id AND o.order_id = e.order_id;

3. 通过 Hint 优化 Lookup 性能

SELECT * FROM orders o /*+ OPTIONS('client.lookup.queue-size'='2560') */
INNER JOIN order_enhance e /*+ OPTIONS('client.lookup.queue-size'='2560') */
ON o.user_id = e.user_id AND o.order_id = e.order_id;

作业运行参数配置

使用 Delta Join 时，需在 Flink 作业的运行参数中添加以下配置：

table.exec.delta-join.cache-enabled: 'true'
table.exec.async-lookup.buffer-capacity: '3000'
table.optimizer.delta-join.strategy: FORCE
table.exec.delta-join.left.cache-size: '100'
table.exec.delta-join.right.cache-size: '1000'

成功启用后，可在 Flink 作业状态总览中看到 Delta Join 节点，表明优化已生效。

功能说明

Fluss 主键表支持作为 Flink SQL 的维表使用，通过 Lookup Join 可以在流处理中实时关联维度数据。Fluss 支持高效的 KV 点查和二级索引查找，适合低延迟的维表关联场景。

使用限制

• 仅支持主键表作为维表。

• Join 条件必须包含维表的全部主键字段。

同步与异步模式

SQL 示例

-- 通过 Hint 配置 Lookup 缓存，进行维表关联
SELECT
  o.order_id,
  o.amount,
  u.user_name,
  p.product_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.cache' = 'PARTIAL', 'lookup.partial-cache.expire-after-write' = '60s', 'lookup.partial-cache.max-rows' = '10000') */
  FOR SYSTEM_TIME AS OF o.proctime AS u
  ON o.user_id = u.user_id
LEFT JOIN `fluss_catalog`.`fluss_db`.product_dim
  FOR SYSTEM_TIME AS OF o.proctime AS p
  ON o.product_id = p.product_id;

-- 使用同步 Lookup 模式（适合调试场景）
SELECT o.order_id, o.amount, d.user_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.async' = 'false') */
  FOR SYSTEM_TIME AS OF o.proctime AS d
  ON o.user_id = d.user_id;

功能说明

Fluss 主键表支持配置不同的 Merge Engine，用于控制相同主键的多条记录如何合并。通过 table.merge-engine 参数设置，支持以下策略：

SQL 示例

-- 使用 first_row 实现去重
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.event_dedup (
    event_id STRING,
    event_type STRING,
    event_time TIMESTAMP(3),
    payload STRING,
    PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'first_row'
);

-- 使用 versioned 按版本列更新
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_profile (
    user_id BIGINT,
    user_name STRING,
    age INT,
    update_version BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'versioned',
  'table.merge-engine.versioned.ver-column' = 'update_version'
);

功能说明

Fluss 支持分区表，通过 PARTITIONED BY 将数据按分区键划分到不同的物理分区中，适用于按时间或业务维度管理数据生命周期和提升查询性能的场景。分区表支持单字段分区和多字段分区，并提供手动管理、自动管理和动态创建三种正交的分区管理策略。

基本规则

• 分区键的数据类型需要是 STRING。

• 主键表的分区键需要是主键的子集。

• 自动分区规则仅在建表时配置，建表后不可修改。

分区管理策略

三种策略是正交的，可以组合使用。例如，可以同时开启自动管理（按时间自动创建和清理）和动态创建（兜底处理未预创建的分区）。

SQL 示例

1. 创建单字段分区表（自动分区管理）

CREATE TABLE `fluss_catalog`.fluss_db.site_access (
  site_id BIGINT,
  user_name STRING,
  pv BIGINT,
  event_day STRING,
  PRIMARY KEY (site_id, event_day) NOT ENFORCED
) PARTITIONED BY (event_day) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-precreate' = '5',
  'table.auto-partition.num-retention' = '30',
  'table.auto-partition.time-zone' = 'Asia/Shanghai'
);

2. 创建多字段分区表

CREATE TABLE `fluss_catalog`.fluss_db.region_access (
  access_id BIGINT,
  user_name STRING,
  event_date STRING,
  region STRING,
  PRIMARY KEY (access_id, event_date, region) NOT ENFORCED
) PARTITIONED BY (event_date, region) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.key' = 'event_date',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-retention' = '7'
);

说明：多字段分区时，需通过 table.auto-partition.key 指定用于自动分区的时间键。多字段分区仅支持自动过期清理，不支持自动预创建（num-precreate 强制为 0）。

3. 手动管理分区

-- 查看所有分区
SHOW PARTITIONS `fluss_catalog`.fluss_db.site_access;

-- 手动添加分区
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  ADD PARTITION (event_day = '20250315');

-- 手动删除分区
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  DROP PARTITION (event_day = '20250101');

4. 动态创建分区

写入数据时，如果目标分区不存在，客户端会自动创建对应分区（默认开启）：

-- 写入数据，若分区 '20250701' 不存在则自动创建
INSERT INTO `fluss_catalog`.fluss_db.site_access
VALUES (1, 'hello', 100, '20250701');

动态创建由客户端参数 client.writer.dynamic-create-partition.enabled（默认 true）控制，受集群级别的 max.partition.num（默认 1000）和 max.bucket.num（默认 128000）限制。

分区下推

流式或批式查询分区表时，Fluss 支持分区下推（分区裁剪），通过 WHERE 条件中的分区键过滤，仅读取匹配的分区数据，减少不必要的 I/O 开销：

-- 仅读取指定日期的分区数据
SELECT * FROM `fluss_catalog`.fluss_db.site_access
WHERE event_day = '20250315';

-- 多字段分区表按 region 过滤
SELECT * FROM `fluss_catalog`.fluss_db.region_access
WHERE region = 'US';

自动分区参数说明

最佳实践

• 选择合适的分区粒度：根据数据量和查询模式选择时间粒度。数据量大、查询时间范围小的场景建议按 DAY 或 HOUR；数据量较小的场景可按 MONTH 或 QUARTER。

• 合理设置保留策略：通过 num-retention 控制历史分区数量，避免分区数过多导致元数据膨胀和集群管理负担增大。

• 组合使用自动管理与动态创建：自动管理负责常规的时间分区创建和清理，动态创建作为兜底机制处理异常或延迟数据。

• 注意集群限制：单表分区数受 max.partition.num 限制（默认 1000），总 Bucket 数受 max.bucket.num 限制（默认 128000），分区数 x bucket.num 不应超过此上限。

Fluss 采用 Apache Arrow 列式存储格式进行日志传输，支持将过滤和投影操作下推到存储层执行，从而减少数据传输量，查询性能可提升高达 10 倍。

这一特性在以下场景中尤为有效：

• 只需读取宽表中部分列（投影下推）。

• 需要根据条件过滤大量数据（谓词下推）。

日志存储格式通过 table.log.format 参数控制，默认为 ARROW。

功能说明

Fluss 支持湖流一体架构，可将 Fluss 中的实时数据自动同步到数据湖（如 Apache Paimon），实现冷热数据自动分层存储，兼顾实时查询的低延迟与历史数据的低存储成本。

关键特性：

• 自动将 Fluss 中的流式数据写入 Paimon 湖存储。

• 冷热数据自动分层，热数据保留在 Fluss 流存储中，冷数据沉降到 Paimon。

• 通过统一的 Flink SQL 查询接口同时访问流存储和湖存储中的数据。

配置参数

功能说明

Fluss 支持在 Flink SQL 的批模式下对表数据进行快速探查，包括 LIMIT 查询、主键点查和 COUNT(*) 统计，无需启动持续运行的流式作业即可快速查看表中的数据。这对于数据开发调试、数据验证和问题排查非常实用。

LIMIT 查询

支持对主键表和日志表执行 LIMIT 查询，快速获取表中的少量样本数据。

-- 从主键表中查看前 10 条数据
SELECT * FROM `fluss_catalog`.`fluss_db`.orders LIMIT 10;

-- 从日志表中查看前 5 条数据
SELECT * FROM `fluss_catalog`.`fluss_db`.events LIMIT 5;

-- 结合投影，仅查看部分列
SELECT order_id, amount, order_time 
FROM `fluss_catalog`.`fluss_db`.orders LIMIT 20;

主键点查

对于主键表，支持通过 WHERE 条件按主键进行精确点查，利用 Fluss 的 KV 存储引擎实现毫秒级响应。

-- 按主键精确查询单条记录
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
WHERE order_id = 1001;

-- 复合主键的点查
SELECT * FROM `fluss_catalog`.`fluss_db`.user_orders
WHERE user_id = 12345 AND order_id = 6789;

COUNT(*) 统计

支持对主键表执行 COUNT(*) 查询，快速获取表中的总记录数。

-- 统计主键表中的总行数
SELECT COUNT(*) FROM `fluss_catalog`.`fluss_db`.orders;