第 2 篇：「Fluss 架构深入」—— CoordinatorServer、TabletServer 与存储引擎

第 2 篇：「Fluss 架构深入」—— CoordinatorServer、TabletServer 与存储引擎

阅读本文你将了解： Fluss 集群的完整架构、CoordinatorServer 和 TabletServer 的职责分工、LogStore 和 KvStore 的存储机制、Tablet/Bucket 如何组织数据、以及远程存储和 ZooKeeper 的角色。

2.1 架构全景

Fluss 集群由四层构成：

┌──────────────────────────────────┐
│ Flink / Spark / 自定义客户端 │
│ (Stream Read/Write, PK Lookup) │
└──────────┬──────────┬─────────────┘
│ │
┌──────────▼──┐ ┌───▼──────────┐
│ Coordinator │ │ TabletServer │
│ Server │ │ x N │
│ │ │ │
│ ┌─────────┐ │ │ ┌───────────┐ │
│ │ 元数据 │ │ │ │ LogStore │ │
│ │ Tablet │ │ │ │ (WAL/流) │ │
│ │ 分配 │ │ │ ├───────────┤ │
│ │ Rebalance│ │ │ │ KvStore │ │
│ └─────────┘ │ │ │ (RocksDB) │ │
└──────┬──────┘ │ └───────────┘ │
│ └───────┬───────┘
┌──────▼──────────────────────┐
│ ZooKeeper │
│ (集群协调 / 元数据存储) │
└─────────────────────────────┘
┌─────────────────────────────┐
│ Remote Storage │
│ (S3 / Iceberg / Paimon / │
│ Lance / RustFS) │
└─────────────────────────────┘

关键设计原则

2.2 CoordinatorServer 源码剖析

CoordinatorServer 是集群的「大脑」。其核心类结构如下：

org.apache.fluss.server.coordinator
├── CoordinatorServer.java # 启动入口
├── CoordinatorService.java # 核心服务逻辑
├── TableManager.java # 表元数据管理
├── TabletAllocator.java # Tablet 分配策略
└── RebalanceCoordinator.java # 重平衡协调

2.2.1 启动流程

// 简化自 org.apache.fluss.server.coordinator.CoordinatorServer
public class CoordinatorServer {

private final Configuration conf;
private final CoordinatorService coordinatorService;

public void start() throws Exception {
// 1. 初始化 ZooKeeper 连接
ZooKeeperClient zkClient = new ZooKeeperClient(conf.get(ZOOKEEPER_ADDRESS));

// 2. 初始化元数据管理器
MetaDataManager metaDataManager = new MetaDataManager(zkClient);

// 3. 初始化 Tablet 分配器
TabletAllocator tabletAllocator = new TabletAllocator(metaDataManager);

// 4. 启动 RPC 服务
RpcServer rpcServer = new RpcServer(conf.get(COORDINATOR_PORT));
rpcServer.start();

// 5. 启动核心协调服务
coordinatorService = new CoordinatorService(
conf, zkClient, metaDataManager, tabletAllocator, rpcServer
);
coordinatorService.start();
}
}

2.2.2 元数据管理

// 简化自 TableManager 的实现逻辑
public class TableManager {

// 核心元数据结构
private final Map<TablePath, TableDescriptor> tables;
private final Map<Long, List<TabletServerInfo>> tabletAssignments;

/**
* 创建表时的核心流程
*/
public void createTable(TablePath tablePath, TableDescriptor descriptor) {
// 1. 校验表 Schema（PK 表必须定义主键）
SchemaValidator.validate(descriptor);

// 2. 计算所需 Tablet 数量
int numBuckets = descriptor.getBucketNum(); // 用户指定或默认
int numPartitions = descriptor.getPartitionKeys().isEmpty()
? 1
: calculatePartitions(descriptor);
int totalTablets = numBuckets * numPartitions;

// 3. 分配 Tablet 到 TabletServer
List<TabletAssignment> assignments = tabletAllocator.allocate(
totalTablets, getAvailableServers()
);

// 4. 向 ZooKeeper 写入元数据
zkClient.createTableNode(tablePath, descriptor, assignments);

// 5. 通知 TabletServer 加载新 Tablet
for (TabletAssignment assignment : assignments) {
notifyServerToLoadTablet(assignment);
}
}
}

2.2.3 Tablet 分配策略

Fluss 的 Tablet 分配采用轮询+容量感知策略：

// 简化分配逻辑
public class TabletAllocator {

public List<TabletAssignment> allocate(int numTablets, List<TabletServerInfo> servers) {
List<TabletAssignment> assignments = new ArrayList<>();

// 按当前负载排序（轻载优先）
servers.sort(Comparator.comparing(TabletServerInfo::getTabletCount));

int serverIndex = 0;
for (int i = 0; i < numTablets; i++) {
TabletServerInfo selected = servers.get(serverIndex % servers.size());
assignments.add(new TabletAssignment(i, selected.getServerId()));
selected.incrementTabletCount();
serverIndex++;
}

return assignments;
}
}

2.2.4 Rebalance 机制

当集群扩缩容时，CoordinatorServer 触发 Rebalance：

触发条件：
├── 新增 TabletServer 节点
├── TabletServer 节点下线
├── Tablet 负载严重不均衡
└── 管理员手动触发

Rebalance 流程：
1. 计算目标分配方案（最优 Tablet 分布）
2. 对比当前分配方案与目标方案的差异
3. 生成迁移计划（最小迁移成本）
4. 分批执行迁移（避免一次性迁移过多 Tablet）
5. 迁移完成后更新 ZooKeeper 元数据

2.3 TabletServer 源码剖析

2.3.1 核心类结构

org.apache.fluss.server.tablet
├── TabletServer.java # 启动入口
├── TabletService.java # Tablet 生命周期管理
├── ReplicaManager.java # 副本管理
├── log/
│ ├── LogTablet.java # Log 存储实现
│ ├── LogSegment.java # 段文件管理
│ └── LogManager.java # Log 整体管理
└── kv/
├── KvTablet.java # KV 存储实现
├── KvManager.java # KV 整体管理
└── rocksdb/
└── RocksDBKv.java # RocksDB 封装

2.3.2 TabletServer 启动与 Tablet 加载

public class TabletServer {

private final TabletService tabletService;
private final ReplicaManager replicaManager;

public void start() {
// 1. 初始化网络层
TransportLayer transport = new TransportLayer(conf);

// 2. 初始化 LogManager（管理所有 LogTablet）
LogManager logManager = new LogManager(conf);

// 3. 初始化 KvManager（管理所有 KvTablet）
KvManager kvManager = new KvManager(conf);

// 4. 初始化副本管理器
replicaManager = new ReplicaManager(logManager, kvManager);

// 5. 向 CoordinatorServer 注册
registerToCoordinator();

// 6. 启动 Tablet 服务
tabletService = new TabletService(replicaManager, transport);
tabletService.start();
}

/**
* 当 CoordinatorServer 通知加载 Tablet 时调用
*/
public void loadTablet(TabletAssignment assignment) {
long tabletId = assignment.getTabletId();
TablePath tablePath = assignment.getTablePath();

// 如果是 Leader 副本，需要同时加载 LogTablet 和 KvTablet
if (assignment.isLeader()) {
// 加载 LogTablet（WAL）
LogTablet logTablet = replicaManager.getOrCreateLogTablet(
tablePath, tabletId
);

// 如果是 PK 表，还需要加载 KvTablet
if (assignment.getTableType() == TableType.PRIMARY_KEY) {
KvTablet kvTablet = replicaManager.getOrCreateKvTablet(
tablePath, tabletId
);
// 可能需要从 LogTablet 重放 WAL 恢复 KvTablet
kvTablet.recoverFromLog(logTablet);
}
}
}
}

2.3.3 LogTablet：日志存储引擎

LogTablet 是 Fluss 写入路径的核心。每个 LogTablet 由多个 LogSegment 组成。

// 简化自 org.apache.fluss.server.tablet.log.LogTablet
public class LogTablet {

private final long tabletId;
private final ConcurrentSkipListMap<Long, LogSegment> segments;
private final ReplicaManager replicaManager;

/**
* 追加消息到 LogTablet
*/
public long append(LogRecordBatch batch) {
// 1. 获取当前活跃的 Segment
LogSegment activeSegment = getActiveSegment();

// 2. 检查 Segment 是否已满（默认 1GB）
if (activeSegment.size() >= maxSegmentSize) {
activeSegment = rollNewSegment();
}

// 3. 写入数据到 .log 文件
long offset = activeSegment.append(batch);

// 4. 更新 .index 文件（稀疏索引：offset → 物理位置）
if (shouldIndex(offset)) {
activeSegment.appendIndex(offset, activeSegment.getPosition());
}

// 5. 如果是 Leader，同步到 Follower 副本（ISR 机制）
if (isLeader()) {
replicaManager.replicateToFollowers(tabletId, offset);
}

return offset;
}
}

LogSegment 物理布局：

LogTablet (tablet_id=0)
├── 00000000000000000000.log # Segment 0: offset 0 ~ 9999
├── 00000000000000000000.index # 稀疏索引（每 4KB 一条）
├── 00000000000000010000.log # Segment 1: offset 10000 ~ 19999
├── 00000000000000010000.index
├── 00000000000000020000.log # Segment 2: offset 20000 ~ …
└── 00000000000000020000.index

2.3.4 KvTablet：RocksDB 封装

PK 表的可更新和可查询能力来源于 KvTablet。

// 简化自 org.apache.fluss.server.tablet.kv.KvTablet
public class KvTablet {

private final long tabletId;
private final RocksDBKv rocksDB;
private final LogTablet walLogTablet;

/**
* 写入（Put/Upsert）操作
*/
public void put(byte[] key, byte[] value) {
// 1. 先写入 WAL（LogTablet），保证持久性
long walOffset = walLogTablet.append(
new LogRecord(key, value, RecordType.PUT)
);

// 2. 写入 RocksDB
rocksDB.put(key, value);
}

/**
* 删除操作
*/
public void delete(byte[] key) {
// 1. WAL 记录删除
walLogTablet.append(new LogRecord(key, null, RecordType.DELETE));

// 2. RocksDB 标记删除
rocksDB.delete(key);
}

/**
* 点查询
*/
public byte[] get(byte[] key) {
// 直接从 RocksDB 读取，亚毫秒级延迟
return rocksDB.get(key);
}

/**
* 故障恢复：从 WAL 重放所有操作到 RocksDB
*/
public void recoverFromLog(LogTablet logTablet) {
LogScanner scanner = logTablet.newScanner(0); // 从 offset 0 开始
LogRecord record;
while ((record = scanner.next()) != null) {
switch (record.getType()) {
case PUT:
rocksDB.put(record.getKey(), record.getValue());
break;
case DELETE:
rocksDB.delete(record.getKey());
break;
}
}
}
}

RocksDB 配置：

// Fluss 对 RocksDB 的优化配置
public class RocksDBConfig {
// 使用 Bloom Filter 加速点查询
public static final boolean BLOOM_FILTER = true; // 10 bits per key

// Block Cache 大小（默认 256MB）
public static final long BLOCK_CACHE_SIZE = 256 * 1024 * 1024;

// LSM 树配置：减少写放大
public static final int WRITE_BUFFER_SIZE = 64 * 1024 * 1024; // 64MB
public static final int MAX_WRITE_BUFFER_NUMBER = 3;

// Compaction 策略：Level Compaction
public static final int NUM_LEVELS = 7;
}

2.4 数据分布层次

Database: "ecommerce"
│
├── Table: "orders"
│ ├── Partition: dt=2026-08-01
│ │ ├── Bucket 0 → Tablet 0 (LogTablet + KvTablet)
│ │ ├── Bucket 1 → Tablet 1 (LogTablet + KvTablet)
│ │ ├── Bucket 2 → Tablet 2 (LogTablet + KvTablet)
│ │ └── Bucket 3 → Tablet 3 (LogTablet + KvTablet)
│ ├── Partition: dt=2026-08-02
│ │ ├── Bucket 0 → Tablet 4
│ │ ├── Bucket 1 → Tablet 5
│ │ └── …
│ └── …
│
├── Table: "user_profiles" (PK Table, 无分区)
│ ├── Bucket 0 → Tablet 100 (LogTablet + KvTablet)
│ ├── Bucket 1 → Tablet 101
│ └── Bucket 2 → Tablet 102

Bucket 分配规则

/**
* 根据 bucketing key 决定数据落入哪个 Bucket
* 默认使用 Hash 分区
*/
public class BucketingFunction {

public int getBucket(byte[] bucketingKey, int numBuckets) {
int hash = MurmurHash3.hash32(bucketingKey);
return Math.abs(hash) % numBuckets;
}
}

同一个 Bucket 的 LogTablet 和 KvTablet 始终分配在同一个 TabletServer 上，以保证：

• 写入路径：先写 LogTablet(WAL) 再写 KvTablet，本地操作无网络延迟
• 恢复路径：KvTablet 从本地 LogTablet 重放 WAL，快速恢复

2.5 副本机制与 ISR

Fluss 的副本机制仅应用于 LogTablet。KvTablet 当前不支持副本（通过 WAL 恢复）。

Tablet 0 (replication factor = 3)
├── Leader → TabletServer-1 (处理读写)
├── Follower → TabletServer-2 (同步复制)
└── Follower → TabletServer-3 (同步复制)

ISR (In-Sync Replica) = { TabletServer-1, TabletServer-2, TabletServer-3 }

写入一致性保证

Client Write
│
▼
Leader (TabletServer-1)
│
├── 1. 写入本地 LogTablet
│
├── 2. 同步复制到 ISR 中的 Follower
│ ├── Follower-1 (TabletServer-2): ACK ✓
│ └── Follower-2 (TabletServer-3): ACK ✓
│
├── 3. 收到所有 ISR 的 ACK（或达到 min.insync.replicas）
│
└── 4. 返回成功给客户端

Leader 故障切换

1. ZooKeeper 检测到 Leader (TabletServer-1) 心跳超时
2. CoordinatorServer 从 ISR 中选择新 Leader
→ 选择 TabletServer-2（log end offset 与旧 Leader 最接近）
3. CoordinatorServer 更新 ZooKeeper 元数据
4. TabletServer-2 晋升为 Leader，开始处理读写
5. 客户端自动重定向到新 Leader

2.6 远程存储层

远程存储基于 S3 兼容接口，核心配置：

# fluss-conf.yaml
remote.data.dir: s3://my–bucket/fluss–data/
s3.endpoint: https://s3.amazonaws.com
s3.access-key: AKIAIOSFODNN7EXAMPLE
s3.secret-key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

分层存储流程

LogTablet Local Disk (SSD)
│ 保留 1-3 天数据
│ 写入：低延迟 < 1ms
│ 读取：热数据直接本地读取
│
▼ Tiering Service 定期执行
│
Remote Storage (S3/对象存储)
│ 保留 30-365 天数据
│ 格式：Parquet（高压缩比）
│ 读取：客户端可直接批量读取

源码支撑： org.apache.fluss.server.tiering.TieringService 负责定期将本地热数据 Compaction 为 Parquet 格式后上传到远程存储。

2.7 总结与下一篇预告

下一篇我们将学习 Fluss 最核心的用户接口：表设计。如何选择 Log Table 还是 PK Table？分区和分桶如何搭配？Schema Evolution 怎么用？这些都是生产环境中每天都在面对的问题。

本文基于 Apache Fluss 0.9.1 源码。项目 GitHub: https://github.com/apache/fluss