像 Git 一样管理数据：Nessie + Iceberg 实现数据版本控制
Tags: #Nessie #Iceberg #DataVersioning #Lakehouse #GitForData #智策平台
Coomia发布于 2025年7月11日19 分钟阅读
分享本文Twitter / X
“系列：S3 数据基座 · 第 2 篇 | 难度：高级 | 阅读时间：20 分钟
像 Git 一样管理数据：Nessie + Iceberg 实现数据版本控制
Tags: #Nessie #Iceberg #DataVersioning #Lakehouse #GitForData #智策平台
#TL;DR
在智策平台中，我们使用 Project Nessie 为 Apache Iceberg 表提供 Git 风格的版本控制：分支（Branch）、标签（Tag）、提交（Commit）、合并（Merge）。每个 World 对应一个 Nessie 分支，Release 对应一个 Tag，数据变更自动生成 Commit 记录。本文详细介绍 Git 概念如何映射到数据管理、三方合并的冲突解决机制、时间旅行查询的实现，以及 WorldManagerService 的核心设计。
#1. 为什么数据需要版本控制
#1.1 传统数据管理的困境
在没有版本控制的数据平台中，常见痛点包括：
Code
传统数据管理：
时间线 ──────────────────────────────►
T1: 数据正确 T2: 某人改错了 T3: 发现问题
┌──────┐ ┌──────┐ ┌──────┐
│ OK │ ──► │ BUG! │ ──► │ ??? │
│ │ 修改 │ │ 发现 │ 无法 │
│ │ │ │ │ 回滚 │
└──────┘ └──────┘ └──────┘
问题：
- 谁改的？ → 不知道
- 改了什么？ → 不知道
- 能回滚吗？ → 不能
- 能并行修改吗？ → 不能
#1.2 Git 模型的启示
Git 在代码管理中解决了同样的问题。核心概念：
| Git 概念 | 含义 | 数据场景映射 |
|---|---|---|
| Repository | 代码仓库 | 数据湖 / 数据仓库 |
| Branch | 分支 | World（业务世界） |
| Commit | 提交 | 数据变更记录 |
| Tag | 标签 | Release（发布版本） |
| Merge | 合并 | 分支数据合并 |
| Diff | 差异 | 数据变更对比 |
| Checkout | 检出 | 切换到某个数据版本 |
#1.3 Nessie + Iceberg 的组合
Code
Nessie + Iceberg 架构：
┌─────────────────────────────────────────┐
│ Nessie Server │
│ ┌─────────────────────────────────┐ │
│ │ Git-Like Catalog API │ │
│ │ - Branch management │ │
│ │ - Commit history │ │
│ │ - Merge operations │ │
│ │ - Tag management │ │
│ └──────────────┬──────────────────┘ │
│ │ │
│ ┌──────────────┴──────────────────┐ │
│ │ Version Store (RocksDB/JDBC) │ │
│ │ - Commit graph │ │
│ │ - Branch pointers │ │
│ │ - Table metadata refs │ │
│ └─────────────────────────────────┘ │
└────────────────────┬────────────────────┘
│ points to
v
┌─────────────────────────────────────────┐
│ Apache Iceberg │
│ ┌─────────────────────────────────┐ │
│ │ Table Metadata │ │
│ │ - Schema evolution │ │
│ │ - Partition spec │ │
│ │ - Snapshot history │ │
│ └──────────────┬──────────────────┘ │
│ │ │
│ ┌──────────────┴──────────────────┐ │
│ │ Data Files (Parquet on MinIO) │ │
│ │ - Columnar storage │ │
│ │ - Statistics per file │ │
│ │ - Delete files (MoR) │ │
│ └─────────────────────────────────┘ │
└─────────────────────────────────────────┘
#2. Git 概念到数据的完整映射
#2.1 Branch = World
在智策平台中，每个 "World"（业务世界）对应 Nessie 的一个分支：
Code
World 与 Branch 映射：
Nessie Branches:
main ──●──●──●──●──●──●──●──●──●──► (生产世界)
│ ▲
│ │ merge
└──●──●──●──┘ (what-if 分析分支)
│
world-whatif-001
world-dev ──●──●──●──●──► (开发世界)
│
└──●──●──► (特性分支)
world-feature-x
Python
# WorldManagerService 中的分支管理
class WorldManagerService:
"""管理 World 到 Nessie Branch 的映射"""
def __init__(self, nessie_client: NessieClient):
self.nessie = nessie_client
async def create_world(
self,
world_id: str,
display_name: str,
source_world_id: str | None = None
) -> World:
"""创建新的 World（Nessie 分支）"""
branch_name = f"world-{world_id}"
if source_world_id:
# 从现有世界派生（类似 git checkout -b）
source_branch = f"world-{source_world_id}"
source_ref = await self.nessie.get_reference(source_branch)
await self.nessie.create_reference(
branch_name=branch_name,
source_hash=source_ref.hash
)
else:
# 从 main 分支创建
main_ref = await self.nessie.get_reference("main")
await self.nessie.create_reference(
branch_name=branch_name,
source_hash=main_ref.hash
)
return World(
id=world_id,
display_name=display_name,
branch_name=branch_name,
created_at=datetime.utcnow()
)
#2.2 Commit = 数据变更记录
每次对 World 中的数据进行修改，都会生成一个 Nessie Commit：
Code
Commit 历史示例：
world-prod-001 分支:
commit: abc123 "添加 500 条设备记录"
│
├── entity_common: +500 rows
└── entity_edge: +1200 rows (设备-部门关系)
commit: def456 "更新设备状态为已维修"
│
└── entity_common: ~50 rows (status: broken→repaired)
commit: ghi789 "删除已退役设备"
│
├── entity_common: -30 rows
└── entity_edge: -75 rows
Python
async def commit_changes(
self,
world_id: str,
operations: list[DataOperation],
message: str,
author: str
) -> CommitResult:
"""提交数据变更到 World 对应的 Nessie 分支"""
branch_name = f"world-{world_id}"
branch_ref = await self.nessie.get_reference(branch_name)
# 构建 Iceberg 操作
iceberg_ops = []
for op in operations:
if op.type == OperationType.INSERT:
iceberg_ops.append(
IcebergAppend(table=op.table, data_files=op.files)
)
elif op.type == OperationType.UPDATE:
iceberg_ops.append(
IcebergOverwrite(
table=op.table,
delete_files=op.old_files,
data_files=op.new_files
)
)
elif op.type == OperationType.DELETE:
iceberg_ops.append(
IcebergDelete(table=op.table, delete_files=op.files)
)
# 提交到 Nessie
result = await self.nessie.commit(
branch=branch_name,
expected_hash=branch_ref.hash,
operations=iceberg_ops,
commit_meta=CommitMeta(
message=message,
author=author,
timestamp=datetime.utcnow()
)
)
return CommitResult(
commit_hash=result.hash,
branch=branch_name,
message=message,
timestamp=result.timestamp
)
#2.3 Tag = Release
Release 版本使用 Nessie Tag 实现不可变的快照引用：
Code
Release 与 Tag 映射：
main ──●──●──●──●──●──●──●──●──●──►
│ │ │
▼ ▼ ▼
tag: tag: tag:
v1.0 v1.1 v2.0
(Q1报表) (修正版) (年报)
Python
async def create_release(
self,
world_id: str,
release_name: str,
description: str
) -> Release:
"""创建 Release（Nessie Tag）"""
branch_name = f"world-{world_id}"
branch_ref = await self.nessie.get_reference(branch_name)
tag_name = f"release-{world_id}-{release_name}"
await self.nessie.create_tag(
tag_name=tag_name,
hash=branch_ref.hash
)
return Release(
name=release_name,
tag=tag_name,
commit_hash=branch_ref.hash,
description=description,
created_at=datetime.utcnow()
)
#3. 三方合并：数据冲突解决
#3.1 合并场景
当两个 World 分支需要合并时（例如将 what-if 分析结果合并到生产世界），可能产生冲突：
Code
三方合并示意：
共同祖先 (Base)
│
┌────┴────┐
│ │
main 分支 what-if 分支
修改了 A 修改了 A (冲突!)
修改了 B 修改了 C (无冲突)
│ │
└────┬────┘
│
合并结果
A: 需要解决冲突
B: 采用 main 的修改
C: 采用 what-if 的修改
#3.2 冲突检测算法
Python
class ThreeWayMerger:
"""三方合并实现"""
async def merge(
self,
source_branch: str,
target_branch: str,
strategy: MergeStrategy = MergeStrategy.NORMAL
) -> MergeResult:
"""执行三方合并"""
# 获取三方引用
source_ref = await self.nessie.get_reference(source_branch)
target_ref = await self.nessie.get_reference(target_branch)
base_ref = await self.nessie.find_common_ancestor(
source_ref.hash, target_ref.hash
)
# 计算差异
source_diff = await self.compute_diff(base_ref.hash, source_ref.hash)
target_diff = await self.compute_diff(base_ref.hash, target_ref.hash)
# 检测冲突
conflicts = self.detect_conflicts(source_diff, target_diff)
if conflicts and strategy == MergeStrategy.NORMAL:
return MergeResult(
status=MergeStatus.CONFLICT,
conflicts=conflicts,
message="Merge conflicts detected"
)
# 无冲突或使用自动解决策略
if strategy == MergeStrategy.THEIRS:
resolved = self.resolve_with_theirs(conflicts)
elif strategy == MergeStrategy.OURS:
resolved = self.resolve_with_ours(conflicts)
else:
resolved = []
# 执行合并
result = await self.nessie.merge(
from_branch=source_branch,
to_branch=target_branch,
expected_hash=target_ref.hash
)
return MergeResult(
status=MergeStatus.SUCCESS,
commit_hash=result.hash,
merged_changes=len(source_diff) + len(target_diff),
resolved_conflicts=len(resolved)
)
def detect_conflicts(
self,
source_diff: list[DiffEntry],
target_diff: list[DiffEntry]
) -> list[Conflict]:
"""检测冲突：同一实体在两个分支上都被修改"""
conflicts = []
source_keys = {d.entity_key for d in source_diff}
target_keys = {d.entity_key for d in target_diff}
overlapping = source_keys & target_keys
for key in overlapping:
source_change = next(d for d in source_diff if d.entity_key == key)
target_change = next(d for d in target_diff if d.entity_key == key)
# 同时修改同一字段才算冲突
if self.has_field_conflict(source_change, target_change):
conflicts.append(Conflict(
entity_key=key,
source_change=source_change,
target_change=target_change
))
return conflicts
#3.3 冲突解决策略
| 策略 | 说明 | 适用场景 |
|---|---|---|
NORMAL | 检测冲突但不自动解决，返回冲突列表 | 需要人工审核的场景 |
OURS | 目标分支（被合并到的分支）优先 | 生产环境优先 |
THEIRS | 来源分支（要合并的分支）优先 | 实验结果采纳 |
FIELD_LEVEL | 字段级合并（无冲突字段自动合并） | 大部分场景推荐 |
Python
# 字段级合并示例
class FieldLevelMerger:
"""字段级别的细粒度合并"""
def merge_entity(
self,
base: dict,
source: dict,
target: dict
) -> tuple[dict, list[str]]:
"""
三方字段级合并
base: 共同祖先版本
source: 来源分支版本
target: 目标分支版本
返回: (合并结果, 冲突字段列表)
"""
merged = dict(base)
conflicts = []
all_fields = set(base.keys()) | set(source.keys()) | set(target.keys())
for field in all_fields:
base_val = base.get(field)
source_val = source.get(field)
target_val = target.get(field)
if source_val == target_val:
# 两边改成一样的，无冲突
merged[field] = source_val
elif source_val == base_val:
# 只有目标分支修改了，采用目标
merged[field] = target_val
elif target_val == base_val:
# 只有来源分支修改了，采用来源
merged[field] = source_val
else:
# 真正的冲突：两边都改了且不同
conflicts.append(field)
merged[field] = target_val # 默认采用目标
return merged, conflicts
#4. 时间旅行查询
#4.1 通过 Commit Hash 查询
SQL
-- 查看某个特定提交时的数据状态
SELECT * FROM entity_common
AT BRANCH 'world-prod-001'
AT COMMIT 'abc123def456'
WHERE object_type_id = 'Equipment'
LIMIT 100;
#4.2 通过时间戳查询
SQL
-- 查看 2026 年 1 月 15 日的数据状态
SELECT * FROM entity_common
AT BRANCH 'world-prod-001'
AS OF TIMESTAMP '2026-01-15T10:00:00Z'
WHERE object_type_id = 'Equipment';
#4.3 Python SDK 实现
Python
class TemporalQueryService:
"""时间旅行查询服务"""
async def query_at_commit(
self,
world_id: str,
commit_hash: str,
query: str
) -> QueryResult:
"""在指定 commit 时间点查询数据"""
branch_name = f"world-{world_id}"
# 获取 commit 对应的 Iceberg snapshot
ref = await self.nessie.get_reference(
branch_name,
hash_on_ref=commit_hash
)
# 获取该时间点的 Iceberg 表元数据
table_metadata = await self.nessie.get_table_metadata(
ref=ref,
table_name="entity_common"
)
# 使用对应 snapshot 执行查询
snapshot_id = table_metadata.current_snapshot_id
return await self.iceberg_engine.query(
table="entity_common",
snapshot_id=snapshot_id,
sql=query
)
async def query_at_timestamp(
self,
world_id: str,
timestamp: datetime,
query: str
) -> QueryResult:
"""在指定时间戳查询数据"""
branch_name = f"world-{world_id}"
# 查找最接近指定时间的 commit
log = await self.nessie.get_commit_log(
branch_name,
filter=f"timestamp <= {timestamp.isoformat()}"
)
if not log.entries:
raise ValueError(f"No commits found before {timestamp}")
# 使用最近的 commit
commit_hash = log.entries[0].hash
return await self.query_at_commit(world_id, commit_hash, query)
#5. Nessie API 详解
#5.1 核心 API
Code
Nessie REST API 端点：
┌─────────────────────────────────────────────────┐
│ Trees API (分支与标签管理) │
├─────────────────────────────────────────────────┤
│ GET /api/v2/trees │
│ POST /api/v2/trees │
│ GET /api/v2/trees/{name} │
│ DELETE /api/v2/trees/{name} │
│ PUT /api/v2/trees/{name} │
├─────────────────────────────────────────────────┤
│ Content API (表内容管理) │
├─────────────────────────────────────────────────┤
│ GET /api/v2/trees/{name}/contents │
│ GET /api/v2/trees/{name}/contents/{key} │
│ POST /api/v2/trees/{name}/contents │
├─────────────────────────────────────────────────┤
│ Commit API (提交管理) │
├─────────────────────────────────────────────────┤
│ GET /api/v2/trees/{name}/history │
│ POST /api/v2/trees/{name}/history/transplant │
│ POST /api/v2/trees/{name}/history/merge │
├─────────────────────────────────────────────────┤
│ Diff API (差异对比) │
├─────────────────────────────────────────────────┤
│ GET /api/v2/trees/{from}/diff/{to} │
└─────────────────────────────────────────────────┘
#5.2 coomia-dip 的 Nessie 客户端封装
Python
class NessieClient:
"""Nessie API 客户端封装"""
def __init__(self, base_url: str, auth_token: str | None = None):
self.base_url = base_url.rstrip("/")
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {auth_token}"} if auth_token else {}
)
# === Branch Operations ===
async def list_branches(self) -> list[Branch]:
"""列出所有分支"""
resp = await self.session.get(
f"{self.base_url}/api/v2/trees",
params={"type": "BRANCH"}
)
data = await resp.json()
return [Branch(**ref) for ref in data["references"]]
async def create_branch(
self, name: str, source_hash: str
) -> Branch:
"""创建新分支"""
resp = await self.session.post(
f"{self.base_url}/api/v2/trees",
json={
"type": "BRANCH",
"name": name,
"hash": source_hash
}
)
return Branch(**(await resp.json()))
async def delete_branch(self, name: str, hash: str) -> None:
"""删除分支"""
await self.session.delete(
f"{self.base_url}/api/v2/trees/{name}",
params={"type": "BRANCH", "expectedHash": hash}
)
# === Commit Operations ===
async def commit(
self,
branch: str,
expected_hash: str,
operations: list[Operation],
commit_meta: CommitMeta
) -> CommitResponse:
"""提交变更"""
resp = await self.session.post(
f"{self.base_url}/api/v2/trees/{branch}/history/commit",
json={
"expectedHash": expected_hash,
"operations": [op.to_dict() for op in operations],
"commitMeta": commit_meta.to_dict()
}
)
return CommitResponse(**(await resp.json()))
async def get_commit_log(
self,
branch: str,
max_records: int = 100,
filter: str | None = None
) -> CommitLog:
"""获取提交历史"""
params = {"maxRecords": max_records}
if filter:
params["filter"] = filter
resp = await self.session.get(
f"{self.base_url}/api/v2/trees/{branch}/history",
params=params
)
return CommitLog(**(await resp.json()))
# === Merge Operations ===
async def merge(
self,
from_branch: str,
to_branch: str,
expected_hash: str
) -> MergeResponse:
"""合并分支"""
from_ref = await self.get_reference(from_branch)
resp = await self.session.post(
f"{self.base_url}/api/v2/trees/{to_branch}/history/merge",
json={
"fromRefName": from_branch,
"fromHash": from_ref.hash,
"expectedHash": expected_hash
}
)
return MergeResponse(**(await resp.json()))
# === Diff Operations ===
async def diff(
self, from_ref: str, to_ref: str
) -> DiffResponse:
"""获取两个引用之间的差异"""
resp = await self.session.get(
f"{self.base_url}/api/v2/trees/{from_ref}/diff/{to_ref}"
)
return DiffResponse(**(await resp.json()))
#6. Iceberg 表管理
#6.1 Iceberg Snapshot 机制
Code
Iceberg 快照链：
Snapshot Chain:
S1 ──► S2 ──► S3 ──► S4 (current)
│ │ │ │
│ │ │ └── manifest-list-4
│ │ │ ├── manifest-a: [file-7, file-8]
│ │ │ └── manifest-b: [file-5, file-6]
│ │ │
│ │ └── manifest-list-3
│ │ └── manifest-a: [file-5, file-6]
│ │
│ └── manifest-list-2
│ └── manifest-a: [file-3, file-4]
│
└── manifest-list-1
└── manifest-a: [file-1, file-2]
MinIO 中的文件布局:
s3://lakehouse/
├── entity_common/
│ ├── metadata/
│ │ ├── v1.metadata.json
│ │ ├── v2.metadata.json
│ │ ├── v3.metadata.json
│ │ └── v4.metadata.json (current)
│ ├── data/
│ │ ├── file-1.parquet
│ │ ├── file-2.parquet
│ │ ├── ...
│ │ └── file-8.parquet
│ └── manifests/
│ ├── manifest-list-1.avro
│ ├── manifest-list-2.avro
│ ├── manifest-list-3.avro
│ ├── manifest-list-4.avro
│ ├── manifest-a.avro
│ └── manifest-b.avro
└── entity_edge/
└── ...
#6.2 Schema Evolution
Python
class IcebergSchemaManager:
"""Iceberg Schema 演化管理"""
async def add_column(
self,
table_name: str,
column_name: str,
column_type: str,
comment: str | None = None
) -> None:
"""添加新列（不影响已有数据）"""
table = self.catalog.load_table(table_name)
with table.update_schema() as update:
update.add_column(column_name, column_type, comment)
async def rename_column(
self,
table_name: str,
old_name: str,
new_name: str
) -> None:
"""重命名列（利用 Iceberg 的 ID 映射）"""
table = self.catalog.load_table(table_name)
with table.update_schema() as update:
update.rename_column(old_name, new_name)
async def evolve_partition(
self,
table_name: str,
new_partition_spec: list[PartitionField]
) -> None:
"""分区演化（不需要重写数据）"""
table = self.catalog.load_table(table_name)
with table.update_spec() as update:
for field in new_partition_spec:
update.add_field(field.source, field.transform, field.name)
#7. WorldManagerService 完整实现
#7.1 服务架构
Code
WorldManagerService 架构：
┌─────────────────────────────────────────┐
│ WorldManagerService │
│ │
│ ┌──────────┐ ┌────────────────────┐ │
│ │ World │ │ Branch │ │
│ │ CRUD │ │ Management │ │
│ └────┬─────┘ └────────┬───────────┘ │
│ │ │ │
│ ┌────┴─────┐ ┌────────┴───────────┐ │
│ │ Release │ │ Merge │ │
│ │ Mgmt │ │ Service │ │
│ └────┬─────┘ └────────┬───────────┘ │
│ │ │ │
│ ┌────┴─────────────────┴───────────┐ │
│ │ NessieClient │ │
│ └──────────────┬───────────────────┘ │
│ │ │
└─────────────────┼───────────────────────┘
│ HTTP/REST
v
┌────────────────┐
│ Nessie Server │
└────────────────┘
#7.2 gRPC 服务定义
PROTOBUF
// world_manager.proto
service WorldManagerService {
// World CRUD
rpc CreateWorld(CreateWorldRequest) returns (WorldResponse);
rpc GetWorld(GetWorldRequest) returns (WorldResponse);
rpc ListWorlds(ListWorldsRequest) returns (ListWorldsResponse);
rpc DeleteWorld(DeleteWorldRequest) returns (Empty);
// Branch operations
rpc ForkWorld(ForkWorldRequest) returns (WorldResponse);
rpc MergeWorld(MergeWorldRequest) returns (MergeResponse);
// Release operations
rpc CreateRelease(CreateReleaseRequest) returns (ReleaseResponse);
rpc ListReleases(ListReleasesRequest) returns (ListReleasesResponse);
// History
rpc GetCommitHistory(CommitHistoryRequest) returns (CommitHistoryResponse);
rpc GetDiff(DiffRequest) returns (DiffResponse);
}
message ForkWorldRequest {
string source_world_id = 1;
string new_world_id = 2;
string display_name = 3;
string description = 4;
}
message MergeWorldRequest {
string source_world_id = 1;
string target_world_id = 2;
MergeStrategy strategy = 3;
string message = 4;
}
enum MergeStrategy {
NORMAL = 0;
OURS = 1;
THEIRS = 2;
FIELD_LEVEL = 3;
}
#8. 实践场景
#8.1 What-If 分析
Code
What-If 分析工作流：
1. 创建分析分支
main ──●──●──●──► (生产数据)
│
└──► world-whatif-001 (分析分支)
2. 在分析分支上修改参数
world-whatif-001: 修改定价模型参数
运行模拟计算
生成预测结果
3. 对比分析结果
diff(main, world-whatif-001)
→ 收入变化: +12%
→ 客户流失风险: -5%
4. 决策：采纳或放弃
如果采纳: merge world-whatif-001 → main
如果放弃: delete world-whatif-001
#8.2 数据审计
Python
async def audit_entity_changes(
self,
world_id: str,
entity_id: str,
start_time: datetime,
end_time: datetime
) -> list[AuditEntry]:
"""审计实体的变更历史"""
branch_name = f"world-{world_id}"
# 获取时间范围内的提交历史
log = await self.nessie.get_commit_log(
branch_name,
filter=f"timestamp >= {start_time.isoformat()} "
f"AND timestamp <= {end_time.isoformat()}"
)
audit_entries = []
for i, entry in enumerate(log.entries):
if i + 1 < len(log.entries):
# 对比相邻提交
diff = await self.nessie.diff(
from_ref=f"{branch_name}@{log.entries[i+1].hash}",
to_ref=f"{branch_name}@{entry.hash}"
)
# 过滤出目标实体的变更
entity_changes = [
d for d in diff.diffs
if d.key.contains(entity_id)
]
if entity_changes:
audit_entries.append(AuditEntry(
commit_hash=entry.hash,
timestamp=entry.timestamp,
author=entry.author,
message=entry.message,
changes=entity_changes
))
return audit_entries
#8.3 数据回滚
Python
async def rollback_world(
self,
world_id: str,
target_commit: str,
reason: str
) -> CommitResult:
"""将 World 回滚到指定提交"""
branch_name = f"world-{world_id}"
current_ref = await self.nessie.get_reference(branch_name)
# 创建回滚分支
rollback_branch = f"{branch_name}-rollback-{target_commit[:8]}"
await self.nessie.create_branch(rollback_branch, target_commit)
# 将回滚分支合并回主分支（使用 THEIRS 策略）
result = await self.nessie.merge(
from_branch=rollback_branch,
to_branch=branch_name,
expected_hash=current_ref.hash
)
# 清理回滚分支
rollback_ref = await self.nessie.get_reference(rollback_branch)
await self.nessie.delete_branch(rollback_branch, rollback_ref.hash)
return CommitResult(
commit_hash=result.hash,
branch=branch_name,
message=f"Rollback to {target_commit[:8]}: {reason}"
)
#9. 性能与运维
#9.1 性能基准
| 操作 | 延迟 | 吞吐量 |
|---|---|---|
| 创建分支 | 5ms | 200/sec |
| 提交（单表） | 15ms | 60/sec |
| 提交（三表联合） | 45ms | 20/sec |
| 合并（无冲突） | 80ms | 10/sec |
| 合并（有冲突检测） | 200ms | 5/sec |
| 获取提交历史（100条） | 30ms | 30/sec |
| Diff（两个分支） | 120ms | 8/sec |
| 时间旅行查询 | +20ms（额外开销） | — |
#9.2 Nessie 存储后端选择
| 后端 | 适用场景 | 性能 | 高可用 |
|---|---|---|---|
| RocksDB | 开发/测试 | 最快 | 单节点 |
| PostgreSQL | 中小规模生产 | 良好 | 主从复制 |
| DynamoDB | AWS 大规模生产 | 良好 | 原生 HA |
| MongoDB | 通用大规模生产 | 良好 | 副本集 |
#9.3 垃圾回收策略
Python
class NessieGarbageCollector:
"""Nessie + Iceberg 垃圾回收"""
async def collect(
self,
max_age_days: int = 30,
dry_run: bool = True
) -> GCResult:
"""清理过期的快照和数据文件"""
# 1. 找出所有活跃引用（分支 + 标签）
active_refs = await self.get_all_active_references()
# 2. 收集所有活跃快照 ID
active_snapshots = set()
for ref in active_refs:
snapshots = await self.get_reachable_snapshots(ref)
active_snapshots.update(snapshots)
# 3. 找出过期快照
all_snapshots = await self.get_all_snapshots()
expired = [
s for s in all_snapshots
if s.id not in active_snapshots
and s.timestamp < datetime.utcnow() - timedelta(days=max_age_days)
]
if dry_run:
return GCResult(
expired_snapshots=len(expired),
reclaimable_bytes=sum(s.size for s in expired),
deleted=False
)
# 4. 删除过期数据文件
for snapshot in expired:
await self.delete_snapshot_files(snapshot)
return GCResult(
expired_snapshots=len(expired),
reclaimed_bytes=sum(s.size for s in expired),
deleted=True
)
#10. 与其他 Layer 的集成
Code
Nessie + Iceberg 在 coomia-dip 中的集成：
┌─────────────┐ ┌──────────────┐
│ Control │ │ Intelligence │
│ Layer (B) │ │ Layer (D) │
│ │ │ │
│ WorldMgr │ │ What-If │
│ Service │ │ Analysis │
└──────┬──────┘ └──────┬───────┘
│ │
│ gRPC │ gRPC
│ │
┌──────┴──────────────────┴───────┐
│ Data Layer (C) │
│ │
│ ┌────────────────────────────┐ │
│ │ NessieIcebergService │ │
│ │ - Branch CRUD │ │
│ │ - Commit & Merge │ │
│ │ - Time Travel │ │
│ │ - Diff │ │
│ └──────────────┬─────────────┘ │
│ │ │
│ ┌──────────────┴─────────────┐ │
│ │ Nessie Server │ │
│ │ + Iceberg Catalog │ │
│ └──────────────┬─────────────┘ │
│ │ │
│ ┌──────────────┴─────────────┐ │
│ │ MinIO (S3 Storage) │ │
│ └────────────────────────────┘ │
└──────────────────────────────────┘
#Key Takeaways
-
数据版本控制不是奢侈品，而是必需品。 在复杂的企业数据平台中，缺乏版本控制意味着无法追溯、无法回滚、无法并行分析。
-
Nessie 提供了完整的 Git 语义。 Branch、Commit、Tag、Merge、Diff——所有 Git 核心操作都有对应的数据操作。
-
三方合并是并行数据分析的基石。 不同团队可以在各自的 World 分支上独立工作，最终通过合并整合结果。
-
时间旅行查询依赖 Iceberg 快照机制。 每个 Commit 对应一个 Iceberg Snapshot，可以精确回到任意历史时间点。
-
World/Branch/Release 的映射设计是核心。 这个映射关系决定了整个平台的数据隔离和版本管理策略。
#Next Article
下一篇 S3-03《DuckDB 嵌入式分析引擎：轻量级计算的秘密武器》 将介绍 DuckDB 如何在智策平台中与 Doris 互补，为函数上下文中的轻量级分析提供零网络延迟的嵌入式计算能力。
Tags: #ProjectNessie #ApacheIceberg #DataVersioning #GitForData #Lakehouse #ThreeWayMerge #TimeTravel #智策平台 #coomia-dip