Skip to content

Commit 08545ca

Browse files
gyang94gyang94
committed
fix: image
1 parent fd028b7 commit 08545ca

File tree

6 files changed

+286
-36
lines changed

6 files changed

+286
-36
lines changed

‎docs/notes/fluss/03-tiering-service-deep-dive-zh.md‎

Lines changed: 194 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,194 @@
1+
---
2+
title: Tiering Service Deep Dive
3+
tags: fluss
4+
outline: deep
5+
---
6+
7+
# Tiering Service Deep Dive
8+
9+
## 背景
10+
11+
![](./img/03/background.png)
12+
13+
Fluss 湖仓架构的核心是**分层服务 (Tiering Service)** —— 这是一个智能的、策略驱动的数据管道，它无缝地连接着您的实时 Fluss 集群和高性价比的湖仓存储。该服务持续从 Fluss 集群摄取新事件，并自动将较旧或访问频率较低的数据迁移到更冷的存储层中，且整个过程不会中断正在进行的查询。通过根据可配置的规则平衡热、温、冷存储，分层服务确保了最新数据立即可查，同时又能经济高效地归档历史记录。在本篇深度剖析中，我们将探讨 Fluss 的分层服务如何编排数据、保障数据一致性，并在兼顾性能与成本的前提下完成数据分析工作。
14+
15+
## Flink 分层服务 (Flink Tiering Service)
16+
17+
Fluss 分层服务是作为一个 Flink job 实现的，它持续地将数据从 Fluss 集群流式传输到您的数据湖中。其执行图非常简单，由三个算子（operators）组成：
18+
19+
```
20+
Source: TieringSource -> TieringCommitter -> Sink: Writer
21+
```
22+
23+
- **TieringSource**: 从 Fluss 表读取记录并将其写入数据湖。
24+
- **TieringCommitter**: 通过推进数据湖仓和 Fluss 集群中的偏移量（offsets）来提交每个同步批次（sync batch）。
25+
- **No-Op Sink**: 一个虚拟接收器（dummy sink），不执行任何实际操作。
26+
27+
在接下来的章节中，我们将深入探讨 TieringSource 和 TieringCommitter，看看它们究竟是如何编排实时存储与历史存储之间的无缝数据传输的。
28+
29+
## TieringSource
30+
31+
![](./img/03/tiering-source.png)
32+
33+
**TieringSource 算子** 从 Fluss 分层表（tiering table）读取记录并将其写入您的数据湖。它构建在 Flink 的 Source V2 API ([FLIP-27](https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface)) 之上，分解为两个核心组件：**TieringSourceEnumerator** 和 **TieringSourceReader**。其高层工作流程如下：
34+
35+
1. **Enumerator** 向 CoordinatorService 查询当前分层表的元数据。
36+
2. 一旦收到表信息，Enumerator 生成“分片”（splits，即数据分区）并将它们分配给 Reader。
37+
3. **Reader** 获取每个分片的实际数据。
38+
4. 随后，Reader 将这些记录写入数据湖。
39+
40+
在接下来的章节中，我们将深入探讨 TieringSourceEnumerator 和 TieringSourceReader 的内部工作机制，了解它们如何实现从 Fluss 到湖仓的可靠、可扩展的数据摄取。
41+
42+
### TieringSourceEnumerator
43+
44+
![](./img/03/tiering-source-enumerator.png)
45+
46+
**TieringSourceEnumerator** 通过五个关键步骤编排分片的创建和分配：
47+
48+
1. **心跳请求 (Heartbeat Request)**：使用 RPC 客户端向 Fluss 服务器发送 **`lakeTieringHeartbeatRequest`**。
49+
2. **心跳响应 (Heartbeat Response)**：接收包含分层表元数据以及已完成、失败和进行中表的同步状态的 **`lakeTieringHeartbeatResponse`**。
50+
3. **湖分层信息 (Lake Tiering Info)**：将返回的 **`lakeTieringInfo`** 转发给 **`TieringSplitGenerator`**。
51+
4. **分片生成 (Split Generation)**：**`TieringSplitGenerator`** 生成一组 **`TieringSplits`**——每个分片代表一个待处理的数据分区。
52+
5. **分片分配 (Split Assignment)**：将这些 **`TieringSplits`** 分配给 **`TieringSourceReader`** 实例，以便后续摄取到数据湖中。
53+
54+
#### RpcClient
55+
56+
- **发送心跳 (Sending Heartbeats)**：它构造并发送 **`LakeTieringHeartbeatRequest`** 消息。该消息携带三个表列表——**`tiering_tables`**（进行中）、**`finished_tables`**（已完成）和 **`failed_tables`**（失败）——以及一个可选的 **`request_table`** 标志（用于请求新的分层工作）。
57+
- **接收响应 (Receiving Responses)**：它等待一个包含以下内容的 **`LakeTieringHeartbeatResponse`** 响应：
58+
- **`coordinator_epoch`**：Coordinator 的当前纪元（epoch）。
59+
- **`tiering_table`** (可选)：一个 **`PbLakeTieringTableInfo`** 消息（包含 **`table_id`**、**`table_path`** 和 **`tiering_epoch`**），描述下一个要进行分层的表。
60+
- **`tiering_table_resp`**、**`finished_table_resp`** 和 **`failed_table_resp`**：反映每个表状态的心跳响应列表。
61+
- **转发元数据 (Forwarding Metadata)**：它解析返回的 **`PbLakeTieringTableInfo`** 和同步状态响应，然后将组装好的 **`lakeTieringInfo`** 转发给 **`TieringSplitGenerator`** 用于创建分片。
62+
63+
64+
#### TieringSplitGenerator
65+
66+
![](./img/03/tiering-split-generator.png)
67+
68+
**TieringSplitGenerator** 计算您的湖仓与 Fluss 集群之间精确的数据差异（delta），然后为每个需要同步的数据段生成 **`TieringSplit`** 任务。它使用 **`FlussAdminClient`** 获取三个核心元数据：
69+
70+
1. **湖快照 (Lake Snapshot)**
71+
- 调用湖元数据 API 获取 **`LakeSnapshot`** 对象，该对象包含：
72+
- **`snapshotId`**（数据湖中最新已提交的快照 ID）
73+
- **`tableBucketsOffset`**（一个映射，将每个 **`TableBucket`** 关联到其在湖仓中的日志偏移量）
74+
2. **当前桶偏移量 (Current Bucket Offsets)**
75+
- 向 Fluss 服务器查询每个桶（bucket）当前的日志结束偏移量（log end offset），捕获输入流的高水位标记（high-water mark）。
76+
3. **KV 快照 (KV Snapshots) (针对主键表)**
77+
- 获取一个 **`KvSnapshots`** 记录，包含：
78+
- **`tableId`** 和可选的 **`partitionId`**
79+
- **`snapshotIds`**（每个桶的最新快照 ID）
80+
- **`logOffsets`**（在该快照之后恢复读取的日志位置）
81+
82+
利用 **`LakeSnapshot`**、实时的桶偏移量以及（如果适用）**`KvSnapshots`**，生成器计算出哪些日志段存在于 Fluss 中但尚未提交到湖仓。然后，它为每个段生成一个 **`TieringSplit`**——每个分片精确定义了要摄取的桶和偏移量范围——从而实现了实时存储与历史存储之间的增量式、高效同步。
83+
84+
#### TieringSplit
85+
86+
**TieringSplit** 抽象精确定义了需要同步的表桶（table bucket）中的哪一部分数据。它捕获三个公共字段：
87+
88+
- **tablePath**: 目标表的完整路径。
89+
- **tableBucket**: 该表中的特定桶（分片/shard）。
90+
- **partitionName** (可选): 分区键（如果表是分区的）。
91+
92+
有两种具体分片类型：
93+
94+
1. **TieringLogSplit** (用于仅追加的“日志”表)
95+
- **startingOffset**: 湖仓中最后提交的日志偏移量。
96+
- **stoppingOffset**: 实时 Fluss 桶中的当前结束偏移量。
97+
- 此分片定义了一个需要摄取的新日志记录的连续范围。
98+
2. **TieringSnapshotSplit** (用于主键表)
99+
- **snapshotId**: Fluss 中最新快照的标识符。
100+
- **logOffsetOfSnapshot**: 拍摄该快照时的日志偏移量。
101+
- 此分片让 TieringSourceReader 能够重放（replay）自该快照以来的所有 CDC（变更数据捕获）事件，确保状态是最新的。
102+
103+
通过将每个表分解成这些定义明确的分片，分层服务能够以增量、可靠且并行的方式，仅同步数据湖中缺失的那部分数据。
104+
105+
### TieringSourceReader
106+
107+
![](./img/03/tiering-source-reader.png)
108+
109+
**TieringSourceReader** 从 Enumerator 拉取分配的分片（splits），使用 **`TieringSplitReader`** 从 Fluss 服务器获取相应的记录，然后将它们写入数据湖。其工作流程分解如下：
110+
111+
1. **分片选择 (Split Selection)**
112+
113+
该 Reader 从其队列中选取一个分配的 **`TieringSplit`**。
114+
115+
2. **Reader 分派 (Reader Dispatch)**
116+
117+
根据分片类型，它实例化对应的 Reader：
118+
119+
- **LogScanner**：用于 **`TieringLogSplit`**（仅追加表）
120+
- **BoundedSplitReader**：用于 **`TieringSnapshotSplit`**（主键表）
121+
3. **数据获取 (Data Fetch)**
122+
123+
选定的 Reader 从 Fluss 服务器获取由分片的偏移量或快照边界定义的记录。
124+
125+
4. **湖写入 (Lake Writing)**
126+
127+
检索到的记录被移交给 Lake Writer，由后者将它们持久化到数据湖中。
128+
129+
130+
通过清晰分离分片分配、Reader 选择、数据获取和湖写入，TieringSourceReader 确保了流数据和快照数据可扩展、并行地摄取到您的湖仓中。
131+
132+
#### LakeWriter & LakeTieringFactory
133+
134+
**LakeWriter** 负责将 Fluss 记录持久化到您的数据湖中，它是通过一个可插拔的 **LakeTieringFactory** 接口实例化的。该接口定义了 Fluss 如何与不同的湖存储格式（例如 Paimon、Iceberg）交互：
135+
136+
137+
```java
138+
public interface LakeTieringFactory {
139+
140+
LakeWriter<WriteResult> createLakeWriter(WriterInitContext writerInitContext);
141+
142+
****SimpleVersionedSerializer<WriteResult> getWriteResultSerializer();
143+
144+
LakeCommitter<WriteResult, CommitableT> createLakeCommitter(
145+
CommitterInitContext committerInitContext);
146+
147+
SimpleVersionedSerializer<CommitableT> getCommitableSerializer();
148+
}
149+
```
150+
151+
- **createLakeWriter(WriterInitContext)**：构建一个 **`LakeWriter`**，用于将 Fluss 行转换为目标表格式。
152+
- **getWriteResultSerializer()**：提供用于序列化 Writer 输出的序列化器。
153+
- **createLakeCommitter(CommitterInitContext)**：构造一个 **`LakeCommitter`**，用于最终确定并原子性地提交数据文件。
154+
- **getCommitableSerializer()**：提供用于可提交令牌（committable tokens）的序列化器。
155+
156+
默认情况下，Fluss 包含一个基于 Paimon 的分层工厂；Iceberg 的支持即将推出。一旦 **`TieringSourceReader`** 通过 **`LakeWriter`** 写入一批记录，它就会将产生的写入元数据向下游发出给 **TieringCommitOperator**，后者随后在湖仓和 Fluss 集群中提交这些更改。
157+
158+
#### Stateless
159+
160+
**`TieringSourceReader`** 被设计为完全无状态——它本身不进行状态检查点（checkpoint）或存储任何 **`TieringSplit`** 信息。相反，每次检查点（checkpoint）只返回一个空列表，将所有分片跟踪工作留给 **`TieringSourceEnumerator`**：
161+
162+
```java
163+
@Override
164+
public List<TieringSplit> snapshotState(long checkpointId) {
165+
// 无状态：Reader 状态中不持有任何分片return Collections.emptyList();
166+
}
167+
```
168+
169+
通过将分片分配完全委托给 Enumerator，Reader 保持轻量级且易于扩展，始终从 Coordinator 处获取新的工作单元。
170+
171+
## TieringCommitter
172+
173+
**TieringCommitter** 算子收集来自 TieringSource 的分层表同步写入结果，然后将结果提交到湖仓和 Fluss 服务器以更新状态。
174+
175+
![](./img/03/tiering-committer.png)
176+
177+
**TieringCommitter** 算子通过获取 TieringSourceReader 输出的 **`WriteResult`** 并分两个阶段提交它们（先提交到数据湖，然后提交回 Fluss）来结束每个同步周期，最后向 Flink coordinator 发出状态事件。它利用两个组件：
178+
179+
- **LakeCommitter**：由可插拔的 **`LakeTieringFactory`** 提供，该组件原子性地将写入的文件提交到湖仓，并返回新的快照 ID。
180+
- **FlussTableLakeSnapshotCommitter**：使用该快照 ID，它更新 Fluss 集群的分层表状态，使 Fluss 服务器和湖仓保持同步。
181+
182+
端到端流程是：
183+
184+
1. **收集写入结果**：从 TieringSourceReader 收集当前检查点的写入结果。
185+
2. **湖仓提交**：通过 **`LakeCommitter`** 完成文件并推进湖仓快照。
186+
3. **Fluss 更新**：使用 **`FlussTableLakeSnapshotCommitter`**，向 Fluss CoordinatorService 确认成功或失败。
187+
4. **事件发出**：向 Flink **`OperatorCoordinator`** 发出 **`FinishedTieringEvent`**（成功或完成时）或 **`FailedTieringEvent`**（出错时）。
188+
189+
TieringCommitter 算子确保了您的实时 Fluss 集群与分析型湖仓之间具有精确一次（exactly-once）语义的一致性同步。
190+
191+
## 结论
192+
193+
在本次深度剖析中，我们拆解了 Fluss 分层服务的每一层——从 TieringSource（Enumerator、RpcClient 和 SplitGenerator）开始，探讨了分片类型和无状态的 TieringSourceReader，并探索了可插拔的 LakeWriter/LakeCommitter 集成。然后，我们了解了 TieringCommitter（及其 LakeCommitter 和 FlussTableLakeSnapshotCommitter）如何确保在您的数据湖和 Fluss 集群之间进行原子性的、精确一次的提交。这些组件共同构建了一个强大的管道，能够可靠地同步实时流和历史快照，为您在实时负载和分析存储之间提供无缝、可扩展的一致性服务。
194+

‎docs/notes/fluss/04-hands-on-fluss-lakehouse.md‎

Lines changed: 91 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -12,17 +12,56 @@ Fluss persists historical data in a lakehouse storage layer while keeping real-t
1212

1313
In this tutorial, we’ll show you how to build a local Fluss lakehouse environment, perform essential data operations, and get hands-on experience with the end-to-end Fluss lakehouse architecture.
1414

15-
## Integrate the Lakehouse Locally
15+
## Integrate the Paimon Lakehouse with S3 Locally
16+
17+
We’ll use **Fluss 0.7** and **Flink 1.20** to run the tiering service on a local cluster, with **Paimon** as the lake format and **S3** as paimon storage. Follow these steps:
18+
19+
### Minio Setup
20+
21+
1. Install Minio object storage locally.
22+
23+
Follow the official ![guide](https://min.io/docs/minio/macos/index.html).
24+
25+
2. Start minio server
26+
27+
Run this command with a local path to store minio data.
28+
```
29+
minio server /tmp/minio-data
30+
```
31+
32+
3. Verify in Minio WebUI.
33+
34+
If minio server is successfully running, there will be several endpoints and an account shown:
35+
36+
```
37+
API: http://192.168.2.236:9000 http://127.0.0.1:9000
38+
RootUser: minioadmin
39+
RootPass: minioadmin
40+
41+
WebUI: http://192.168.2.236:61832 http://127.0.0.1:61832
42+
RootUser: minioadmin
43+
RootPass: minioadmin
44+
```
45+
Open the webUI link and login with the user account.
46+
47+
4. Create a `fluss` bucket through webUI.
48+
49+
![](./img/04/fluss-bucket.png)
1650

17-
We’ll use **Fluss 0.7** and **Flink 1.20** to run the tiering service on a local cluster, with **Paimon** as the lake format. Follow these steps:
1851

1952
### Fluss Cluster Setup
2053

2154
1. Download Fluss
2255

23-
Get the Fluss 0.7 binary release from the [official site](https://alibaba.github.io/fluss-docs/downloads/).
56+
Get the Fluss 0.7 binary release from the ![official site](https://alibaba.github.io/fluss-docs/downloads/).
57+
58+
2. Add dependency
59+
60+
Download the `fluss-fs-s3-0.7.0.jar` from Fluss ![official site](https://alibaba.github.io/fluss-docs/downloads/) and put it into `<FLUSS_HOME>/lib`.
2461

25-
2. Configure the Data Lake
62+
Download the `paimon-s3-1.0.1.jar` from Paimon ![official site]() and put it into `<FLUSS_HOME>/plugins/paimon`.
63+
64+
3. Configure the Data Lake
2665

2766
Edit `<FLUSS_HOME>/conf/server.yaml` and add:
2867

@@ -32,9 +71,15 @@ remote.data.dir: /tmp/fluss-remote-data
3271

3372
datalake.format: paimon
3473
datalake.paimon.metastore: filesystem
35-
datalake.paimon.warehouse: /tmp/fluss-paimon-data
74+
datalake.paimon.warehouse: s3://fluss/data
75+
datalake.paimon.s3.endpoint: http://localhost:9000
76+
datalake.paimon.s3.access-key: minioadmin
77+
datalake.paimon.s3.secret-key: minioadmin
78+
datalake.paimon.s3.path.style.access: true
3679
```
3780

81+
Set Paimon as the datalake format and s3 as the warehouse.
82+
3883
3. Start Fluss
3984

4085
```java
@@ -55,17 +100,16 @@ Download `fluss-flink-1.20-0.7.0.jar` from the [Fluss site](https://alibaba.gith
55100
<FLINK_HOME>/lib
56101
```
57102

58-
3. Add Paimon Support
103+
3. Add Paimon Dependencies
59104

60-
- Download `paimon-flink-1.20-1.01.jar` from the [Paimon project site](https://paimon.apache.org/docs/1.0/project/download/) into `<FLINK_HOME>/lib`.
105+
- Download `paimon-flink-1.20-1.0.1.jar` and `paimon-s3-1.0.1.jar` from the [Paimon project site](https://paimon.apache.org/docs/1.0/project/download/) into `<FLINK_HOME>/lib`.
61106
- Copy the Paimon plugin jars from Fluss into `<FLINK_HOME>/lib` .
62107

63108
```java
64109
<FLUSS_HOME>/plugins/paimon/fluss-lake-paimon-0.7.0.jar
65110
<FLUSS_HOME>/plugins/paimon/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
66111
```
67112

68-
69113
4. Increase Task Slots
70114

71115
Edit `<FLINK_HOME>/conf/config.yaml`:
@@ -227,39 +271,50 @@ The __offset and __timestamp columns for these two records are not the default v
227271

228272
6. Inspect the Paimon Files
229273

230-
On your local filesystem, you can verify the Parquet files and manifest under `/tmp/fluss-paimon-data` .
274+
Open the Minio webUI, you will see there exists paimon files in the bucket.
275+
276+
![](./img/04/fluss-bucket-data.png)
277+
278+
In your local filesystem, you can verify the Parquet files and manifest under `/tmp/minio-data` .
231279

232280
```
233-
/tmp/fluss-paimon-data ❯ tree .  00:20:07
281+
/tmp/minio-data ❯ tree .
234282
.
235-
├── default.db
236-
└── fluss.db
237-
└── t_user
238-
├── bucket-0
239-
│ ├── changelog-bd4303c8-80f1-4b7b-9dc6-c6a3ce96aa47-0.parquet
240-
│ ├── changelog-e0209369-16ba-462c-af0a-815f57d72553-0.parquet
241-
│ ├── data-bd4303c8-80f1-4b7b-9dc6-c6a3ce96aa47-1.parquet
242-
│ └── data-e0209369-16ba-462c-af0a-815f57d72553-1.parquet
243-
├── manifest
244-
│ ├── manifest-bbb24f9d-a4b9-4ea6-82ef-97b16c7f23d8-0
245-
│ ├── manifest-bbb24f9d-a4b9-4ea6-82ef-97b16c7f23d8-1
246-
│ ├── manifest-e7bf8240-a442-40f3-8b82-8ff5619dc7e9-0
247-
│ ├── manifest-e7bf8240-a442-40f3-8b82-8ff5619dc7e9-1
248-
│ ├── manifest-list-8e23b5ce-0b39-407f-b8ba-7aaaee629154-0
249-
│ ├── manifest-list-8e23b5ce-0b39-407f-b8ba-7aaaee629154-1
250-
│ ├── manifest-list-8e23b5ce-0b39-407f-b8ba-7aaaee629154-2
251-
│ ├── manifest-list-e10b824d-1a07-47a2-8f91-b12c170cfc43-0
252-
│ ├── manifest-list-e10b824d-1a07-47a2-8f91-b12c170cfc43-1
253-
│ └── manifest-list-e10b824d-1a07-47a2-8f91-b12c170cfc43-2
254-
├── schema
255-
│ └── schema-0
256-
└── snapshot
257-
├── LATEST
258-
├── snapshot-1
259-
└── snapshot-2
283+
└── fluss
284+
└── data
285+
├── default.db__XLDIR__
286+
│   └── xl.meta
287+
└── fluss.db
288+
└── t_user
289+
├── bucket-0
290+
│   ├── changelog-ad07a5a0-10c8-48a9-93c5-11d43e35c9d9-0.parquet
291+
│   │   └── xl.meta
292+
│   └── data-ad07a5a0-10c8-48a9-93c5-11d43e35c9d9-1.parquet
293+
│   └── xl.meta
294+
├── manifest
295+
│   ├── manifest-6bb801ab-9a80-4df1-927f-063b73764cf8-0
296+
│   │   └── xl.meta
297+
│   ├── manifest-6bb801ab-9a80-4df1-927f-063b73764cf8-1
298+
│   │   └── xl.meta
299+
│   ├── manifest-list-e8b9cde3-5b94-456e-a702-cedd498e1c5f-0
300+
│   │   └── xl.meta
301+
│   ├── manifest-list-e8b9cde3-5b94-456e-a702-cedd498e1c5f-1
302+
│   │   └── xl.meta
303+
│   └── manifest-list-e8b9cde3-5b94-456e-a702-cedd498e1c5f-2
304+
│   └── xl.meta
305+
├── schema
306+
│   └── schema-0
307+
│   └── xl.meta
308+
└── snapshot
309+
├── LATEST
310+
│   └── xl.meta
311+
└── snapshot-1
312+
└── xl.meta
313+
314+
20 directories, 11 files
260315
```
261316

262-
7. View Snapshots
317+
1. View Snapshots
263318

264319
Users can also check the snapshots from the system table, by appending `$lake$snapshots` after thefluss table name.
265320

‎docs/notes/fluss/fluss-category.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,4 +12,5 @@ Fluss项目于Flink Forward Asia 2024大会上开源，当前的学习资料不
1212
* [搭建Fluss本地开发环境](./01-development-env-setup.md)
1313
* [Fluss Catalog](./02-fluss-catalog.md)
1414
* [Tiering Service Deep Dive](./03-tiering-service-deep-dive.md)
15+
* [Tiering Service Deep Dive (ZH)](./03-tiering-service-deep-dive-zh.md)
1516
* [Hands-on Fluss Lakehouse](./04-hands-on-fluss-lakehouse.md)

‎docs/notes/fluss/img/03/tiering-split-generator.png‎

3.25 KB
Loading

‎docs/notes/fluss/img/04/fluss-bucket-data.png‎

43.6 KB
Loading

‎docs/notes/fluss/img/04/fluss-bucket.png‎

409 KB
Loading

0 commit comments

Comments
 (0)