事件溯源：审计、血缘追踪与状态重放
大多数系统使用 CRUD 模型管理数据：创建、读取、更新、删除。这在简单场景下完全够用。但当你面对以下需求时，CRUD 就力不从心了：
Coomia发布于 2025年12月17日13 分钟阅读
分享本文Twitter / X
事件溯源：审计、血缘追踪与状态重放
“系列：S10 设计模式 · 第 3 篇 | 难度：高级 | 阅读时间：18 分钟
#TL;DR
- 事件溯源（Event Sourcing）不存储当前状态，而是存储导致当前状态的所有事件。当前状态通过重放事件序列推导而来。
- 在 coomia-dip 中，事件溯源是审计合规、数据血缘追踪和状态回放的基础设施——每一个 Ontology 对象的变更、每一次 Action 的执行、每一条推理链的触发，都以不可变事件记录。
- 通过结合 Nessie（Git-like 版本控制）和 Apache Iceberg（时间旅行查询），coomia-dip 实现了从元数据到数据的全链路可溯源。
#引言：为什么 CRUD 不够用
大多数系统使用 CRUD 模型管理数据：创建、读取、更新、删除。这在简单场景下完全够用。但当你面对以下需求时，CRUD 就力不从心了：
Code
监管审计：3 年前这笔交易的风险评分是怎么算出来的？
故障排查：昨天 15:30 系统为什么拒绝了这个审批？
合规回溯：当时适用的规则版本是什么？
数据修正：发现一个月前的数据有误，能否精确修正并重新计算影响？
CRUD 只告诉你"现在是什么"，不告诉你"为什么变成这样"。你需要翻日志、查备份、问同事，最终可能还是搞不清楚。
事件溯源从根本上解决这个问题：不存储结果，存储过程。
#一、事件溯源的核心概念
#1.1 事件（Event）
事件是已经发生的事实，不可修改，不可删除：
Python
@dataclass(frozen=True)
class DomainEvent:
"""Immutable domain event."""
event_id: str # 全局唯一 ID
event_type: str # 事件类型
aggregate_id: str # 聚合根 ID
aggregate_type: str # 聚合根类型
sequence_number: int # 聚合内序列号
timestamp: datetime # 事件发生时间
payload: dict # 事件数据
metadata: EventMetadata # 元数据
causation_id: str | None = None # 因果链 ID
correlation_id: str | None = None # 关联 ID
@dataclass(frozen=True)
class EventMetadata:
"""Event metadata for audit and lineage."""
actor_id: str # 操作者
actor_type: str # user / system / rule / agent
tenant_id: str # 租户
world_id: str # World 上下文
source_plane: str # 来源 Layer
trace_id: str # 分布式追踪 ID
rule_version: str | None = None # 触发规则的版本
model_version: str | None = None # 推理模型版本
#1.2 事件存储（Event Store）
事件存储是追加写入（append-only）的日志，coomia-dip 使用 Apache Kafka 作为事件总线，Apache Iceberg 作为持久化存储：
Python
class EventStore:
"""Append-only event store backed by Kafka + Iceberg."""
async def append(
self,
events: list[DomainEvent],
expected_version: int | None = None,
) -> None:
"""Append events with optimistic concurrency control."""
if expected_version is not None:
current = await self._get_current_version(
events[0].aggregate_id
)
if current != expected_version:
raise ConcurrencyConflict(
f"Expected version {expected_version}, "
f"got {current}"
)
for event in events:
# 写入 Kafka（实时消费）
await self.kafka_producer.send(
topic=f"events.{event.aggregate_type}",
key=event.aggregate_id,
value=self.serializer.serialize(event),
)
# 同步写入 Iceberg（持久化 + 时间旅行）
await self.iceberg_writer.write_events(events)
async def load_events(
self,
aggregate_id: str,
from_version: int = 0,
to_version: int | None = None,
as_of: datetime | None = None,
) -> list[DomainEvent]:
"""Load events for an aggregate, optionally at a point in time."""
if as_of:
# 时间旅行查询 via Iceberg
return await self.iceberg_reader.read_events(
aggregate_id=aggregate_id,
snapshot_time=as_of,
)
return await self.iceberg_reader.read_events(
aggregate_id=aggregate_id,
from_version=from_version,
to_version=to_version,
)
#1.3 聚合重建
从事件序列重建聚合的当前状态：
Python
class AggregateRepository:
"""Reconstruct aggregates from event streams."""
async def load(
self,
aggregate_id: str,
as_of: datetime | None = None,
) -> Aggregate:
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
as_of=as_of,
)
if not events:
raise AggregateNotFound(aggregate_id)
aggregate = self.aggregate_factory.create_empty(
events[0].aggregate_type
)
for event in events:
aggregate.apply(event)
return aggregate
#二、coomia-dip 中的事件溯源实践
#2.1 Ontology 变更事件
每一次对 Ontology 对象的操作都产生事件：
Python
class OntologyEventTypes:
"""All ontology-level events."""
# Schema 变更
OBJECT_TYPE_CREATED = "ontology.object_type.created"
OBJECT_TYPE_PROPERTY_ADDED = "ontology.object_type.property_added"
OBJECT_TYPE_PROPERTY_MODIFIED = "ontology.object_type.property_modified"
RELATION_TYPE_CREATED = "ontology.relation_type.created"
ACTION_TYPE_REGISTERED = "ontology.action_type.registered"
# 实例变更
OBJECT_CREATED = "ontology.object.created"
OBJECT_UPDATED = "ontology.object.updated"
OBJECT_DELETED = "ontology.object.deleted"
RELATION_CREATED = "ontology.relation.created"
RELATION_DELETED = "ontology.relation.deleted"
# Action 执行
ACTION_SUBMITTED = "ontology.action.submitted"
ACTION_APPROVED = "ontology.action.approved"
ACTION_EXECUTED = "ontology.action.executed"
ACTION_ROLLED_BACK = "ontology.action.rolled_back"
# 推理与决策
REASONING_TRIGGERED = "reasoning.chain.triggered"
REASONING_COMPLETED = "reasoning.chain.completed"
DECISION_MADE = "decision.made"
RULE_FIRED = "rule.fired"
#2.2 Action 执行的完整事件流
一个 Action（如"审批贷款申请"）在事件溯源下的完整生命周期：
Python
class LoanApprovalActionHandler:
"""Handle loan approval with full event sourcing."""
async def handle(self, command: ApproveLoanCommand) -> ActionResult:
events = []
# 1. 提交事件
events.append(DomainEvent(
event_type="action.loan_approval.submitted",
aggregate_id=command.loan_id,
payload={
"applicant_id": command.applicant_id,
"amount": command.amount,
"officer_id": command.officer_id,
},
metadata=self._build_metadata(command),
))
# 2. 规则评估事件
risk_result = await self.risk_engine.evaluate(command.loan_id)
events.append(DomainEvent(
event_type="rule.credit_risk.evaluated",
aggregate_id=command.loan_id,
payload={
"risk_score": risk_result.score,
"risk_factors": risk_result.factors,
"rule_version": risk_result.rule_version,
"model_version": risk_result.model_version,
},
))
# 3. 推理链事件
reasoning = await self.reasoning_engine.run_chain(
chain="loan_assessment",
context={"loan_id": command.loan_id},
)
events.append(DomainEvent(
event_type="reasoning.loan_assessment.completed",
aggregate_id=command.loan_id,
payload={
"recommendation": reasoning.recommendation,
"confidence": reasoning.confidence,
"reasoning_trace": reasoning.trace,
},
))
# 4. 决策事件
decision = self._make_decision(risk_result, reasoning)
events.append(DomainEvent(
event_type="decision.loan_approval.made",
aggregate_id=command.loan_id,
payload={
"decision": decision.outcome,
"approved_amount": decision.approved_amount,
"conditions": decision.conditions,
},
))
# 原子性写入所有事件
await self.event_store.append(events)
return ActionResult(decision=decision, events=events)
#三、审计与合规
#3.1 审计查询
事件溯源让审计查询变得简单：
Python
class AuditService:
"""Audit service powered by event sourcing."""
async def get_audit_trail(
self,
entity_id: str,
start_time: datetime | None = None,
end_time: datetime | None = None,
actor_filter: str | None = None,
) -> list[AuditEntry]:
"""Get complete audit trail for any entity."""
events = await self.event_store.load_events(
aggregate_id=entity_id,
)
entries = []
for event in events:
if start_time and event.timestamp < start_time:
continue
if end_time and event.timestamp > end_time:
continue
if actor_filter and event.metadata.actor_id != actor_filter:
continue
entries.append(AuditEntry(
timestamp=event.timestamp,
action=event.event_type,
actor=event.metadata.actor_id,
actor_type=event.metadata.actor_type,
details=event.payload,
trace_id=event.metadata.trace_id,
))
return entries
async def reconstruct_state_at(
self,
entity_id: str,
point_in_time: datetime,
) -> dict:
"""Reconstruct exact state at any point in time."""
aggregate = await self.aggregate_repo.load(
aggregate_id=entity_id,
as_of=point_in_time,
)
return aggregate.to_dict()
#3.2 合规报告生成
Python
class ComplianceReporter:
"""Generate compliance reports from event streams."""
async def generate_decision_report(
self,
decision_id: str,
) -> ComplianceReport:
"""Generate full decision audit for regulators."""
# 加载决策的完整事件链
events = await self.event_store.load_events(
aggregate_id=decision_id
)
# 按因果链分组
causal_chain = self._build_causal_chain(events)
# 提取每一步的规则版本、模型版本、数据快照
steps = []
for event in causal_chain:
step = ComplianceStep(
timestamp=event.timestamp,
action=event.event_type,
rule_version=event.metadata.rule_version,
model_version=event.metadata.model_version,
input_data=event.payload.get("input"),
output_data=event.payload.get("output"),
actor=event.metadata.actor_id,
)
steps.append(step)
return ComplianceReport(
decision_id=decision_id,
steps=steps,
generated_at=datetime.utcnow(),
reproducible=True, # 可以重放验证
)
#四、数据血缘追踪
#4.1 血缘图构建
事件溯源自然提供了数据血缘信息：
Python
class LineageTracker:
"""Track data lineage from event streams."""
async def build_lineage_graph(
self,
entity_id: str,
depth: int = 5,
) -> LineageGraph:
"""Build lineage graph showing data provenance."""
graph = LineageGraph()
visited = set()
await self._traverse_lineage(
entity_id, graph, visited, depth
)
return graph
async def _traverse_lineage(
self,
entity_id: str,
graph: LineageGraph,
visited: set,
remaining_depth: int,
) -> None:
if entity_id in visited or remaining_depth <= 0:
return
visited.add(entity_id)
events = await self.event_store.load_events(
aggregate_id=entity_id
)
for event in events:
# 每个事件都是血缘图中的一个节点
node = LineageNode(
entity_id=entity_id,
event_type=event.event_type,
timestamp=event.timestamp,
source=event.metadata.source_plane,
)
graph.add_node(node)
# 追踪因果链上游
if event.causation_id:
causation_event = await self.event_store.get_event(
event.causation_id
)
if causation_event:
graph.add_edge(
from_id=causation_event.aggregate_id,
to_id=entity_id,
relation="caused_by",
event_type=causation_event.event_type,
)
await self._traverse_lineage(
causation_event.aggregate_id,
graph, visited, remaining_depth - 1,
)
#4.2 影响分析
当数据源发生变化时，追踪下游影响：
Python
class ImpactAnalyzer:
"""Analyze downstream impact of data changes."""
async def analyze_impact(
self,
source_entity_id: str,
change_type: str,
) -> ImpactReport:
"""Find all entities affected by a change."""
affected = []
queue = [(source_entity_id, 0)]
visited = set()
while queue:
entity_id, depth = queue.pop(0)
if entity_id in visited:
continue
visited.add(entity_id)
# 查找所有引用此实体的事件
downstream = await self.event_store.find_events_referencing(
entity_id=entity_id
)
for event in downstream:
impact = ImpactEntry(
affected_entity=event.aggregate_id,
affected_type=event.aggregate_type,
relationship=event.event_type,
depth=depth,
)
affected.append(impact)
queue.append((event.aggregate_id, depth + 1))
return ImpactReport(
source=source_entity_id,
change_type=change_type,
affected_entities=affected,
total_impact=len(affected),
)
#五、状态重放与时间旅行
#5.1 全量重放
从零开始重建任意时间点的系统状态：
Python
class StateReplayer:
"""Replay events to reconstruct system state."""
async def replay_to(
self,
target_time: datetime,
entity_filter: str | None = None,
) -> ReplayResult:
"""Replay all events up to a target timestamp."""
stream = self.event_store.stream_all_events(
until=target_time,
entity_filter=entity_filter,
)
state = {}
event_count = 0
async for event in stream:
aggregate_id = event.aggregate_id
if aggregate_id not in state:
state[aggregate_id] = self.aggregate_factory.create_empty(
event.aggregate_type
)
state[aggregate_id].apply(event)
event_count += 1
if event_count % 10000 == 0:
self.progress_reporter.report(
f"Replayed {event_count} events"
)
return ReplayResult(
target_time=target_time,
entities_rebuilt=len(state),
events_replayed=event_count,
state=state,
)
#5.2 部分重放（What-If 分析）
修改某些事件后重放，模拟"如果当时做了不同的决定会怎样"：
Python
class WhatIfAnalyzer:
"""Perform what-if analysis by modifying and replaying events."""
async def analyze(
self,
entity_id: str,
modifications: list[EventModification],
) -> WhatIfResult:
"""Replay with modified events to see alternative outcomes."""
original_events = await self.event_store.load_events(
aggregate_id=entity_id
)
# 应用修改
modified_events = self._apply_modifications(
original_events, modifications
)
# 在沙箱中重放修改后的事件
original_state = self._replay_events(original_events)
modified_state = self._replay_events(modified_events)
# 比较差异
diff = self._compute_diff(original_state, modified_state)
return WhatIfResult(
original_outcome=original_state.to_dict(),
modified_outcome=modified_state.to_dict(),
differences=diff,
affected_downstream=await self._trace_downstream_impact(
entity_id, diff
),
)
#六、与 Nessie 的集成
#6.1 Schema 版本化
coomia-dip 使用 Nessie 实现 Schema 的 Git-like 版本管理，与事件溯源互补：
Python
class NessieEventIntegration:
"""Integrate Nessie version control with event sourcing."""
async def record_schema_change(
self,
schema_change: SchemaChange,
) -> None:
# 1. 在 Nessie 中提交 Schema 变更
nessie_commit = await self.nessie_client.commit(
branch=schema_change.branch,
operations=[
NessieOp.put(
key=schema_change.schema_key,
content=schema_change.new_schema,
)
],
message=schema_change.description,
)
# 2. 同时记录事件（关联 Nessie commit hash）
event = DomainEvent(
event_type="ontology.schema.changed",
aggregate_id=schema_change.object_type_id,
payload={
"change_type": schema_change.change_type,
"old_schema": schema_change.old_schema,
"new_schema": schema_change.new_schema,
"nessie_commit": nessie_commit.hash,
"nessie_branch": schema_change.branch,
},
)
await self.event_store.append([event])
#6.2 跨版本查询
结合 Nessie 分支和事件溯源实现跨版本查询：
Python
class CrossVersionQuery:
"""Query data across different schema versions."""
async def query_at_version(
self,
object_type: str,
nessie_ref: str, # branch 或 commit hash
filters: dict,
) -> list[dict]:
"""Query objects using the schema at a specific version."""
# 获取该版本的 Schema
schema = await self.nessie_client.get_content(
key=f"schemas/{object_type}",
ref=nessie_ref,
)
# 使用该 Schema 解释事件
events = await self.event_store.load_events(
aggregate_type=object_type,
as_of=await self._ref_to_timestamp(nessie_ref),
)
# 按旧版 Schema 重建状态
objects = self._rebuild_with_schema(events, schema)
return self._apply_filters(objects, filters)
#七、性能优化
#7.1 快照（Snapshot）
当事件数量很大时，每次从头重放太慢。快照解决这个问题：
Python
class SnapshotManager:
"""Manage aggregate snapshots for performance."""
SNAPSHOT_INTERVAL = 100 # 每 100 个事件创建快照
async def load_with_snapshot(
self,
aggregate_id: str,
) -> Aggregate:
# 1. 查找最新快照
snapshot = await self.snapshot_store.get_latest(aggregate_id)
if snapshot:
# 从快照开始，只重放后续事件
aggregate = self.aggregate_factory.from_snapshot(snapshot)
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
from_version=snapshot.version + 1,
)
else:
aggregate = self.aggregate_factory.create_empty()
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
)
for event in events:
aggregate.apply(event)
# 检查是否需要创建新快照
if aggregate.version - (snapshot.version if snapshot else 0) >= self.SNAPSHOT_INTERVAL:
await self._create_snapshot(aggregate)
return aggregate
async def _create_snapshot(self, aggregate: Aggregate) -> None:
snapshot = AggregateSnapshot(
aggregate_id=aggregate.id,
aggregate_type=aggregate.type,
version=aggregate.version,
state=aggregate.serialize(),
created_at=datetime.utcnow(),
)
await self.snapshot_store.save(snapshot)
#7.2 事件压缩
对于高频更新的聚合，可以在保留完整历史的同时压缩事件流：
Python
class EventCompactor:
"""Compact event streams while preserving full history."""
async def compact(
self,
aggregate_id: str,
before: datetime,
) -> CompactionResult:
"""Compact old events into summary events."""
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
)
# 分为保留区和压缩区
to_compact = [e for e in events if e.timestamp < before]
to_keep = [e for e in events if e.timestamp >= before]
if len(to_compact) < 50:
return CompactionResult(compacted=False)
# 创建摘要事件（保留审计关键信息）
summary = self._create_summary_event(to_compact)
# 将原始事件移入冷存储
await self.cold_storage.archive(to_compact)
# 用摘要事件替换
await self.event_store.replace_compacted(
aggregate_id=aggregate_id,
summary_event=summary,
compacted_count=len(to_compact),
)
return CompactionResult(
compacted=True,
original_count=len(to_compact),
archived=True,
)
#八、CQRS 分离
#8.1 命令侧与查询侧
事件溯源通常与 CQRS（命令查询责任分离）配合使用：
Python
class CommandSide:
"""Handle commands by producing events."""
async def handle_command(self, command: Command) -> list[DomainEvent]:
aggregate = await self.repo.load(command.aggregate_id)
events = aggregate.handle(command)
await self.event_store.append(events)
return events
class QuerySide:
"""Handle queries from read-optimized projections."""
async def handle_query(self, query: Query) -> QueryResult:
# 从物化视图读取，不从事件流
return await self.read_store.query(
entity_type=query.entity_type,
filters=query.filters,
sort=query.sort,
page=query.page,
)
class ProjectionBuilder:
"""Build read-optimized projections from events."""
async def process_event(self, event: DomainEvent) -> None:
handler = self.handlers.get(event.event_type)
if handler:
await handler(event)
async def _handle_object_created(self, event: DomainEvent) -> None:
await self.read_store.upsert(
entity_type=event.aggregate_type,
entity_id=event.aggregate_id,
data=event.payload,
version=event.sequence_number,
)
#九、实战案例：贷款审批全链路溯源
#9.1 场景
监管机构要求银行解释 3 年前一笔贷款拒绝的完整决策过程。
#9.2 溯源过程
Python
# 1. 加载贷款的完整事件历史
events = await audit_service.get_audit_trail(
entity_id="loan-2023-08-15-0042",
start_time=datetime(2023, 8, 14),
end_time=datetime(2023, 8, 16),
)
# 2. 重建当时的状态
state_at_decision = await audit_service.reconstruct_state_at(
entity_id="loan-2023-08-15-0042",
point_in_time=datetime(2023, 8, 15, 14, 30),
)
# 3. 查看当时使用的规则版本
# → rule_version: "credit-risk-v2.3.1"
# → model_version: "xgboost-credit-2023Q3"
# 4. 使用相同版本的规则和模型重放
replay = await what_if.analyze(
entity_id="loan-2023-08-15-0042",
modifications=[], # 无修改，纯重放验证
)
# 5. 确认结果一致 → 决策可解释、可复现
assert replay.original_outcome == replay.modified_outcome
#十、总结
#Key Takeaways
- 事件是事实，状态是推导 —— 事件不可变、不可删除，当前状态通过重放事件推导。
- 审计天然就在那里 —— 事件溯源系统不需要额外的审计日志，事件流本身就是完整的审计轨迹。
- 血缘追踪是因果链的自然延伸 —— 通过 causation_id 和 correlation_id，数据血缘图自动构建。
- 时间旅行是杀手级特性 —— 结合 Iceberg 和 Nessie，可以查询任意时间点的数据和 Schema。
- CQRS 是必要的伴侣 —— 事件溯源的读取性能需要通过物化投影来优化。
- 快照和压缩保证长期可用性 —— 没有优化措施的事件溯源在数据量大时会退化。
#参考资料
- Martin Fowler, Event Sourcing↗
- Greg Young, CQRS and Event Sourcing↗
- Apache Iceberg Time Travel↗
- Project Nessie↗
- coomia-dip 架构总览
tags: event-sourcing audit lineage replay cqrs coomia-dip
下一篇：S10-04 Saga 模式实战