数据流全景：一条数据从采集到决策的完整旅程
要真正理解一个平台的架构，最好的方式不是看静态的架构图，而是跟踪一条数据的完整生命周期。
Coomia发布于 2025年7月2日18 分钟阅读
分享本文Twitter / X
数据流全景：一条数据从采集到决策的完整旅程
“系列：S2 架构全景 · 第 9 篇 | 难度：中级 | 阅读时间：18 分钟
#TL;DR
- 智策平台中一条数据变更从外部数据库到最终决策执行需要经过 12 个阶段，涉及 3 个进程（onto-control、onto-data、onto-intelligence）和 4 种中间件（Kafka、Doris、Iceberg、Temporal）。
- 数据流采用 事件驱动架构，通过 Kafka 解耦各阶段，任何一个阶段的故障不会阻塞上游，保证了系统的弹性和可恢复性。
- 从数据变更到决策执行的端到端延迟目标是 < 5 秒（P99），其中 CDC 捕获 < 1s、Kafka 传输 < 100ms、规则评估 < 500ms、决策执行 < 2s。
#引言：跟踪一条数据的完整旅程
要真正理解一个平台的架构，最好的方式不是看静态的架构图，而是跟踪一条数据的完整生命周期。
想象这样一个场景：一个供应链管理系统中，某个仓库的库存数量从 1000 降到了 50。这个变更会触发一系列连锁反应：
- 库存数据被捕获
- 数据同步到平台
- Ontology 对象被更新
- 派生属性（库存周转率）自动重算
- 规则引擎检测到"库存低于安全阈值"
- 决策引擎选择最优补货方案
- Action 引擎执行采购订单创建
- 审计日志记录全过程
本文将逐步追踪这条数据的旅程，从源头到终点，展示智策平台每一层的处理逻辑。
#1. 全景架构图
Code
数据流全景
=========
┌─────────────┐
│ 外部数据库 │ Stage 1: 数据源变更
│ (MySQL/PG) │
└──────┬──────┘
│ binlog/WAL
▼
┌─────────────┐
│ Flink CDC │ Stage 2: 变更捕获
│ Connector │
└──────┬──────┘
│ CDC Event
▼
┌─────────────┐
│ Kafka │ Stage 3: 事件传输
│ (Topic: │
│ cdc.raw) │
└──────┬──────┘
│
┌────┴────┐
▼ ▼
┌────────┐ ┌────────┐
│ Doris │ │Iceberg │ Stage 4: 数据存储
│ (实时) │ │(历史) │
└────┬───┘ └────────┘
│
▼
┌──────────────┐
│onto-data │ Stage 5: Ontology Runtime
│(Object Store) │ 对象实例更新
└──────┬───────┘
│ 变更事件
▼
┌──────────────┐
│ Kafka │ Stage 6: 变更订阅
│ (Topic: │
│ onto.changes)│
└──────┬───────┘
│
┌────┴────┬──────────┐
▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐
│派生 │ │订阅 │ │物化 │ Stage 7: 级联处理
│属性 │ │通知 │ │视图 │
│重算 │ │ │ │更新 │
└──┬───┘ └──────┘ └──────┘
│
▼
┌──────────────┐
│onto-intelli │ Stage 8: 规则评估
│(Rule Engine) │
└──────┬───────┘
│ 规则匹配
▼
┌──────────────┐
│Decision │ Stage 9: 决策选择
│Engine │
└──────┬───────┘
│ 决策结果
▼
┌──────────────┐
│Action Engine │ Stage 10: Action 执行
│(Temporal) │
└──────┬───────┘
│
┌────┴────┐
▼ ▼
┌──────┐ ┌──────┐
│外部 │ │审计 │ Stage 11-12: 执行 + 审计
│系统 │ │日志 │
└──────┘ └──────┘
#2. Stage 1-2：数据源变更与 CDC 捕获
#2.1 场景设定
SQL
-- 外部系统中的库存变更
UPDATE inventory SET quantity = 50
WHERE warehouse_id = 'WH-001'
AND product_id = 'PROD-ABC';
-- 这条 UPDATE 会生成 binlog 记录：
-- {
-- "table": "inventory",
-- "type": "UPDATE",
-- "before": {"warehouse_id": "WH-001", "product_id": "PROD-ABC", "quantity": 1000},
-- "after": {"warehouse_id": "WH-001", "product_id": "PROD-ABC", "quantity": 50},
-- "ts_ms": 1711234567000
-- }
#2.2 Flink CDC 捕获
Flink CDC Connector 监听外部数据库的 binlog（MySQL）或 WAL（PostgreSQL），将变更事件标准化为统一格式：
Java
// Flink CDC Job 配置
public class InventoryCDCJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
MySqlSource<String> source = MySqlSource.<String>builder()
.hostname("external-db.example.com")
.port(3306)
.databaseList("supply_chain")
.tableList("supply_chain.inventory")
.username("cdc_reader")
.password("***")
.deserializer(new JsonDebeziumDeserializationSchema())
.build();
env.fromSource(source,
WatermarkStrategy.noWatermarks(),
"inventory-cdc")
.map(new CDCEventTransformer()) // 转换为平台标准格式
.sinkTo(KafkaSink.<String>builder()
.setBootstrapServers("kafka:9092")
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setTopic("cdc.raw.inventory")
.setValueSerializationSchema(
new SimpleStringSchema())
.build())
.build());
env.execute("Inventory CDC Pipeline");
}
}
#2.3 CDC 事件标准化
原始 Debezium 事件被转换为平台标准的 CDC Event 格式：
JSON
{
"event_id": "evt-cdc-001-1711234567",
"event_type": "CDC_UPDATE",
"source": {
"connector": "mysql-inventory",
"database": "supply_chain",
"table": "inventory",
"server_id": "db-001"
},
"before": {
"warehouse_id": "WH-001",
"product_id": "PROD-ABC",
"quantity": 1000,
"updated_at": "2026-03-23T10:00:00Z"
},
"after": {
"warehouse_id": "WH-001",
"product_id": "PROD-ABC",
"quantity": 50,
"updated_at": "2026-03-24T10:30:00Z"
},
"op": "u",
"ts_ms": 1711234567000,
"ontology_mapping": {
"object_type": "InventoryRecord",
"primary_key": "WH-001::PROD-ABC",
"world_id": "world-prod"
}
}
#3. Stage 3-4：Kafka 传输与数据存储
#3.1 Kafka Topic 设计
Code
Kafka Topic 架构（三级 Topic）：
第一级：原始 CDC 事件
cdc.raw.inventory -- 库存 CDC
cdc.raw.orders -- 订单 CDC
cdc.raw.shipments -- 物流 CDC
第二级：Ontology 变更事件
onto.changes.objects -- 对象实例变更
onto.changes.links -- 关系变更
onto.changes.properties -- 属性变更
第三级：领域事件
domain.alerts -- 告警事件
domain.decisions -- 决策事件
domain.actions -- Action 执行事件
分区策略：
cdc.raw.* -- 按 primary_key hash 分区（保证同一实体顺序）
onto.changes.* -- 按 world_id + object_type hash 分区
domain.* -- 按 event_type hash 分区
#3.2 双写：Doris（实时查询）+ Iceberg（历史版本）
Code
双写架构：
Kafka Consumer Group 1: Kafka Consumer Group 2:
onto-data (Doris Writer) onto-data (Iceberg Writer)
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Doris │ │ Iceberg │
│ │ │ (on Nessie) │
│ - 实时查询 │ │ │
│ - 最新状态 │ │ - 历史版本 │
│ - 聚合分析 │ │ - 时间旅行 │
│ - 毫秒级响应 │ │ - 审计追溯 │
└─────────────┘ └─────────────┘
写入模式：
Doris: Upsert（更新最新值）
Iceberg: Append（追加历史记录）+ Nessie 版本标签
#3.3 Doris 写入
Java
// onto-data 的 Doris Writer
@Service
public class DorisObjectWriter {
@Autowired
private DorisTemplate dorisTemplate;
public void upsert(CDCEvent event) {
String tableName = resolveTable(event);
Map<String, Object> row = new HashMap<>();
row.put("pk", event.getPrimaryKey());
row.put("world_id", event.getWorldId());
// 将 CDC after 字段映射到 Doris 列
for (Map.Entry<String, Object> entry :
event.getAfter().entrySet()) {
String columnName = toDorisColumn(
event.getObjectType(), entry.getKey()
);
row.put(columnName, entry.getValue());
}
row.put("_updated_at", event.getTimestamp());
row.put("_version", event.getVersion());
dorisTemplate.upsert(tableName, row);
}
}
#3.4 Iceberg + Nessie 写入
Java
// onto-data 的 Iceberg Writer（带 Nessie 版本控制）
@Service
public class IcebergObjectWriter {
@Autowired
private NessieCatalog catalog;
public void appendWithVersion(CDCEvent event) {
TableIdentifier tableId = TableIdentifier.of(
event.getWorldId(),
event.getObjectType()
);
Table table = catalog.loadTable(tableId);
// 构建 Iceberg Record
GenericRecord record = GenericRecord.create(table.schema());
record.setField("pk", event.getPrimaryKey());
record.setField("op", event.getOperation());
record.setField("before_json",
toJson(event.getBefore()));
record.setField("after_json",
toJson(event.getAfter()));
record.setField("ts_ms", event.getTimestamp());
record.setField("event_id", event.getEventId());
// Append 到 Iceberg 表
DataFile dataFile = writeToParquet(table, record);
table.newAppend()
.appendFile(dataFile)
.commit();
// Nessie 打版本标签（可选，按策略触发）
if (shouldTagVersion(event)) {
catalog.commitMeta()
.tag("v-" + event.getTimestamp())
.message("CDC: " + event.getObjectType()
+ " updated")
.commit();
}
}
}
#4. Stage 5-6：Ontology Runtime 与变更订阅
#4.1 对象实例更新
当 CDC 数据写入 Doris 后，onto-data 的 Ontology Runtime 层将其映射为 Ontology 对象实例：
Java
// Ontology Runtime 对象更新
@Service
public class OntologyRuntimeService {
@Autowired
private ObjectTypeRepository objectTypeRepo;
@Autowired
private ChangeEventPublisher eventPublisher;
public void processObjectUpdate(CDCEvent cdcEvent) {
// 1. 查找 ObjectType 定义
ObjectType objectType = objectTypeRepo
.findByApiName(cdcEvent.getObjectType());
// 2. 验证属性映射
Map<String, Object> properties =
mapProperties(objectType, cdcEvent.getAfter());
// 3. 构建 Ontology 变更事件
ObjectChangeEvent changeEvent = ObjectChangeEvent.builder()
.eventId(generateEventId())
.worldId(cdcEvent.getWorldId())
.objectType(objectType.getApiName())
.objectPrimaryKey(cdcEvent.getPrimaryKey())
.changeType(ChangeType.UPDATE)
.beforeProperties(
mapProperties(objectType, cdcEvent.getBefore()))
.afterProperties(properties)
.changedProperties(
detectChangedProperties(
cdcEvent.getBefore(), cdcEvent.getAfter()))
.sourceEventId(cdcEvent.getEventId())
.timestamp(Instant.now())
.build();
// 4. 发布到 Kafka onto.changes.objects
eventPublisher.publish(
"onto.changes.objects", changeEvent);
}
}
#4.2 变更事件格式
JSON
{
"event_id": "onto-chg-001",
"world_id": "world-prod",
"object_type": "InventoryRecord",
"object_primary_key": "WH-001::PROD-ABC",
"change_type": "UPDATE",
"changed_properties": ["quantity"],
"before_properties": {
"warehouseId": "WH-001",
"productId": "PROD-ABC",
"quantity": 1000
},
"after_properties": {
"warehouseId": "WH-001",
"productId": "PROD-ABC",
"quantity": 50
},
"source_event_id": "evt-cdc-001-1711234567",
"timestamp": "2026-03-24T10:30:01Z"
}
#4.3 订阅机制
变更事件发布到 Kafka 后，多个消费者同时订阅：
Code
onto.changes.objects Topic 的消费者：
Consumer Group 1: derived-property-calculator
→ 检查是否有派生属性依赖该属性变更
→ 如果有，触发派生属性重算
Consumer Group 2: subscription-notifier
→ 检查是否有用户/应用订阅了该对象类型的变更
→ 如果有，发送通知（WebSocket/Webhook）
Consumer Group 3: materialized-view-updater
→ 检查是否有物化视图包含该对象
→ 如果有，触发增量更新
Consumer Group 4: rule-engine-trigger
→ 检查是否有规则监听该属性变更
→ 如果有，触发规则评估
#5. Stage 7：派生属性级联重算
#5.1 派生属性依赖 DAG
当 quantity 属性变更时，需要检查哪些派生属性依赖它：
Code
派生属性依赖图 (DAG)：
quantity (基础属性)
│
├──► inventoryValue (派生属性)
│ = quantity × unitPrice
│ │
│ └──► totalWarehouseValue (派生属性)
│ = SUM(inventoryValue) GROUP BY warehouseId
│
└──► stockLevel (派生属性)
= CASE WHEN quantity < safetyThreshold
THEN 'CRITICAL'
WHEN quantity < reorderPoint
THEN 'LOW'
ELSE 'NORMAL' END
│
└──► needsReorder (派生属性)
= stockLevel IN ('CRITICAL', 'LOW')
#5.2 DAG 遍历与重算
Python
# onto-intelligence 的派生属性重算服务
class DerivedPropertyCalculator:
"""派生属性级联重算"""
def __init__(self, dag_store, property_store):
self.dag_store = dag_store
self.property_store = property_store
async def on_property_change(
self, change_event: ObjectChangeEvent
):
"""处理属性变更，触发派生属性重算"""
for changed_prop in change_event.changed_properties:
# 1. 查找依赖该属性的派生属性（拓扑排序）
dependents = self.dag_store.get_dependents_topo_sorted(
object_type=change_event.object_type,
property_name=changed_prop,
)
# 2. 按拓扑顺序重算（保证依赖关系正确）
for derived_prop in dependents:
new_value = await self._calculate(
derived_prop,
change_event.object_primary_key,
change_event.after_properties,
)
# 3. 更新派生属性值
await self.property_store.update_derived(
world_id=change_event.world_id,
object_type=change_event.object_type,
object_pk=change_event.object_primary_key,
property_name=derived_prop.name,
value=new_value,
)
# 4. 发布派生属性变更事件（可能触发下一级级联）
await self._publish_derived_change(
change_event, derived_prop.name, new_value
)
async def _calculate(
self, derived_prop, object_pk, current_properties
):
"""计算单个派生属性的值"""
expression = derived_prop.expression
# 构建计算上下文
context = {**current_properties}
# 如果是聚合类型，需要查询多个对象
if derived_prop.is_aggregation:
context["_aggregation_data"] = (
await self._fetch_aggregation_data(
derived_prop, object_pk
)
)
return expression.evaluate(context)
#5.3 重算结果
Code
重算链路：
quantity: 1000 → 50
inventoryValue: 1000 × $25 = $25,000 → 50 × $25 = $1,250
stockLevel: 'NORMAL' → 'CRITICAL' (safetyThreshold = 100)
needsReorder: false → true
totalWarehouseValue: $500,000 → $476,250
#6. Stage 8-9：规则评估与决策选择
#6.1 规则引擎触发
当 stockLevel 变更为 CRITICAL 时，规则引擎被触发：
Python
# 规则定义（存储在 Ontology 中）
rule_definition = {
"rule_id": "RULE-INV-001",
"name": "低库存自动补货规则",
"trigger": {
"object_type": "InventoryRecord",
"property": "stockLevel",
"condition": "value == 'CRITICAL'"
},
"conditions": [
{
"type": "property_check",
"object_type": "InventoryRecord",
"property": "autoReorderEnabled",
"operator": "==",
"value": True
},
{
"type": "time_check",
"constraint": "business_hours" # 只在工作时间触发
}
],
"actions": [
{
"type": "trigger_decision",
"decision_id": "DEC-REORDER-001"
}
]
}
#6.2 规则评估过程
Python
class RuleEvaluationEngine:
"""规则评估引擎"""
async def evaluate(
self, change_event: ObjectChangeEvent
) -> list[RuleMatch]:
"""评估所有匹配的规则"""
# 1. 查找监听该属性变更的规则
candidate_rules = await self.rule_store.find_rules(
object_type=change_event.object_type,
changed_properties=change_event.changed_properties,
)
matches = []
for rule in candidate_rules:
# 2. 检查触发条件
if not self._check_trigger(
rule.trigger, change_event
):
continue
# 3. 检查附加条件
all_conditions_met = True
for condition in rule.conditions:
if not await self._evaluate_condition(
condition, change_event
):
all_conditions_met = False
break
if all_conditions_met:
matches.append(RuleMatch(
rule=rule,
trigger_event=change_event,
matched_at=datetime.utcnow(),
))
# 4. 处理规则冲突（优先级排序）
resolved = self._resolve_conflicts(matches)
# 5. 执行规则动作
for match in resolved:
await self._execute_actions(match)
return resolved
#6.3 决策引擎
规则匹配后触发决策引擎，选择最优补货方案：
Python
class DecisionEngine:
"""决策引擎"""
async def execute_decision(
self,
decision_id: str,
context: dict,
) -> DecisionResult:
"""执行决策"""
decision = await self.decision_store.get(decision_id)
# 决策树评估
if decision.type == "DECISION_TREE":
return await self._evaluate_tree(
decision.tree, context
)
# 评分模型
elif decision.type == "SCORING_MODEL":
return await self._evaluate_scoring(
decision.model, context
)
async def _evaluate_tree(self, tree, context):
"""决策树评估"""
node = tree.root
while not node.is_leaf:
# 获取评估属性的值
value = await self._resolve_value(
node.property_ref, context
)
# 选择分支
for branch in node.branches:
if branch.condition.evaluate(value):
node = branch.target
break
return DecisionResult(
action=node.action,
confidence=node.confidence,
reasoning_path=self._get_path(tree.root, node),
)
决策结果示例：
JSON
{
"decision_id": "DEC-REORDER-001",
"result": {
"action": "CREATE_PURCHASE_ORDER",
"parameters": {
"supplier_id": "SUP-BEST-001",
"product_id": "PROD-ABC",
"quantity": 500,
"priority": "HIGH",
"delivery_method": "EXPRESS"
},
"confidence": 0.92,
"reasoning_path": [
"stockLevel == CRITICAL → 需要紧急补货",
"supplier_SUP-BEST-001.leadTime < 3days → 选择最快供应商",
"quantity = safetyThreshold × 5 = 500 → 补到安全库存的5倍"
]
}
}
#7. Stage 10-11：Action 执行
#7.1 Action 引擎（Temporal 工作流）
决策结果转化为 Action 执行，通过 Temporal 工作流引擎保证可靠执行：
Python
# Action 执行工作流
@workflow.defn
class PurchaseOrderWorkflow:
"""采购订单创建工作流"""
@workflow.run
async def run(
self, params: PurchaseOrderParams
) -> ActionResult:
# Step 1: 审批检查（金额 > 10万需要人工审批）
if params.estimated_cost > 100000:
approval = await workflow.execute_activity(
request_approval,
args=[params],
start_to_close_timeout=timedelta(hours=24),
)
if not approval.approved:
return ActionResult(
status="REJECTED",
reason=approval.reason,
)
# Step 2: 创建采购订单（调用外部 ERP）
try:
po_result = await workflow.execute_activity(
create_purchase_order_in_erp,
args=[params],
start_to_close_timeout=timedelta(minutes=5),
retry_policy=RetryPolicy(
maximum_attempts=3,
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
),
)
except Exception as e:
# 补偿逻辑：通知相关人员
await workflow.execute_activity(
notify_po_creation_failure,
args=[params, str(e)],
start_to_close_timeout=timedelta(minutes=1),
)
raise
# Step 3: 更新 Ontology 对象状态
await workflow.execute_activity(
update_inventory_reorder_status,
args=[params.product_id, po_result.po_id],
start_to_close_timeout=timedelta(minutes=1),
)
# Step 4: 发送通知
await workflow.execute_activity(
send_notification,
args=[
f"采购订单 {po_result.po_id} 已创建",
params.notify_users,
],
start_to_close_timeout=timedelta(minutes=1),
)
return ActionResult(
status="COMPLETED",
po_id=po_result.po_id,
details=po_result,
)
#7.2 Action 执行时序
Code
Action 执行时序：
Time onto-intelligence Temporal External ERP
│
│ 决策结果 ───────────► 启动工作流
│ │
│ 审批检查
│ (自动通过,
│ 金额 < 10万)
│ │
│ Activity:
│ create_po ──────────► POST /api/po
│ │ │
│ │ 创建成功
│ │ ◄──────────────── PO-2026-001
│ │
│ Activity:
│ update_onto
│ ◄───────────────────── gRPC Update
│ 更新对象状态
│ │
│ Activity:
│ notify
│ │
│ 工作流完成
│
#8. Stage 12：审计日志
#8.1 全链路审计
每个阶段都会产生审计记录，最终汇总到审计日志服务：
Python
# 审计日志记录
class AuditLogger:
"""全链路审计日志"""
async def log_full_chain(
self, chain_id: str, events: list[AuditEvent]
):
"""记录完整的数据到决策链路"""
audit_record = {
"chain_id": chain_id,
"start_time": events[0].timestamp,
"end_time": events[-1].timestamp,
"total_duration_ms": (
events[-1].timestamp - events[0].timestamp
).total_seconds() * 1000,
"stages": [
{
"stage": "CDC_CAPTURE",
"event_id": "evt-cdc-001",
"timestamp": "2026-03-24T10:30:00.100Z",
"source": "mysql-inventory",
"duration_ms": 50,
},
{
"stage": "KAFKA_TRANSPORT",
"event_id": "kafka-msg-001",
"timestamp": "2026-03-24T10:30:00.150Z",
"topic": "cdc.raw.inventory",
"partition": 3,
"duration_ms": 80,
},
{
"stage": "DORIS_WRITE",
"timestamp": "2026-03-24T10:30:00.230Z",
"table": "inventory_record",
"duration_ms": 20,
},
{
"stage": "ONTOLOGY_UPDATE",
"timestamp": "2026-03-24T10:30:00.300Z",
"object_type": "InventoryRecord",
"object_pk": "WH-001::PROD-ABC",
"duration_ms": 100,
},
{
"stage": "DERIVED_PROPERTY_CALC",
"timestamp": "2026-03-24T10:30:00.500Z",
"properties_recalculated": [
"inventoryValue",
"stockLevel",
"needsReorder",
],
"duration_ms": 200,
},
{
"stage": "RULE_EVALUATION",
"timestamp": "2026-03-24T10:30:00.800Z",
"rule_id": "RULE-INV-001",
"result": "MATCHED",
"duration_ms": 150,
},
{
"stage": "DECISION_EXECUTION",
"timestamp": "2026-03-24T10:30:01.000Z",
"decision_id": "DEC-REORDER-001",
"result": "CREATE_PURCHASE_ORDER",
"confidence": 0.92,
"duration_ms": 300,
},
{
"stage": "ACTION_EXECUTION",
"timestamp": "2026-03-24T10:30:01.500Z",
"action": "PurchaseOrderWorkflow",
"result": "COMPLETED",
"po_id": "PO-2026-001",
"duration_ms": 2500,
},
],
}
await self.audit_store.save(audit_record)
#8.2 端到端延迟分析
Code
端到端延迟分解（P99 目标）：
Stage 目标延迟 实际延迟
─────────────────────────────────────────────
CDC 捕获 < 1000ms ~100ms
Kafka 传输 < 100ms ~80ms
Doris 写入 < 50ms ~20ms
Ontology 更新 < 200ms ~100ms
派生属性重算 < 500ms ~200ms
规则评估 < 500ms ~150ms
决策执行 < 500ms ~300ms
Action 执行 < 3000ms ~2500ms
─────────────────────────────────────────────
合计 < 5850ms ~3450ms
P99 端到端延迟目标：< 5 秒
实际 P99 延迟：~3.5 秒
#9. 异常场景处理
#9.1 各阶段故障处理
Code
故障处理矩阵：
阶段 故障类型 处理策略
─────────────────────────────────────────────────────
CDC 捕获 数据库连接断开 自动重连 + 从 binlog 位点恢复
Kafka 传输 Broker 不可用 Producer 缓冲 + 重试
Doris 写入 写入超时 DLQ + 告警 + 人工重放
Iceberg 写入 Nessie 冲突 乐观重试（3次）
Ontology 更新 对象类型不存在 记录到 DLQ + 通知管理员
派生属性重算 表达式计算错误 标记属性为 ERROR 状态 + 告警
规则评估 规则冲突 按优先级排序 + 日志记录
决策执行 模型超时 降级到默认决策
Action 执行 外部系统失败 Temporal 自动重试 + 补偿
审计日志 写入失败 本地缓冲 + 异步重试
#9.2 Dead Letter Queue (DLQ) 设计
Code
DLQ 架构：
正常处理流程：
cdc.raw.* ──► Consumer ──► 处理成功 ──► 下一阶段
异常处理流程：
cdc.raw.* ──► Consumer ──► 处理失败
│
▼ (重试3次后)
dlq.cdc.raw.*
│
▼
DLQ Monitor
(每分钟扫描)
│
┌──────────┴──────────┐
▼ ▼
自动重放 告警 + 人工介入
(可恢复错误) (不可恢复错误)
#10. 性能优化策略
#10.1 批量处理
Code
批量优化：
场景：同一个表的大量 CDC 事件（如批量库存更新）
优化前：逐条处理
CDC Event 1 → Kafka → Doris Write → Onto Update → ...
CDC Event 2 → Kafka → Doris Write → Onto Update → ...
...
1000 条 × 3.5s = 3500s（不可接受）
优化后：微批处理
CDC Events 1-100 → Kafka (batch) → Doris Batch Write →
Onto Batch Update → 派生属性批量重算
批量大小: 100 条/批
批量等待: 最多 500ms
1000 条 = 10 批 × 500ms = 5s（可接受）
#10.2 派生属性重算优化
Code
优化策略：
1. 变更合并：500ms 窗口内同一对象的多次变更合并为一次重算
2. DAG 剪枝：只重算实际受影响的派生属性分支
3. 缓存中间结果：相同 DAG 路径的中间计算结果缓存
4. 并行计算：DAG 中无依赖关系的派生属性并行重算
#Key Takeaways
-
事件驱动架构是数据密集型平台的最佳选择：通过 Kafka 解耦 12 个处理阶段，每个阶段独立伸缩、独立故障隔离，保证了系统的弹性和可观测性。
-
Ontology 是数据流的语义枢纽：原始 CDC 事件经过 Ontology 映射后获得业务语义，后续的派生属性、规则、决策都基于 Ontology 语义而非原始数据字段进行操作。
-
端到端可追溯性是智能决策平台的生命线：从数据变更到决策执行的每一步都有审计记录，chain_id 串联完整链路，既满足合规要求也为调试提供了完整视图。
#下一篇预告
S2-10 一致性模型：分布式系统中的数据一致性设计 —— 数据流全景展示了数据如何流动，但没有回答一个关键问题：当多个 World 并行修改、多个消费者并行处理时，如何保证数据一致性？下一篇将深入讨论 Nessie 乐观并发、Kafka 事件顺序保证、World 隔离、以及无分布式事务的一致性方案。
tags: Data-Flow, CDC, Kafka, Doris, Iceberg, Ontology-Runtime, Derived-Property, Rule-Engine, Decision-Engine, Temporal, Audit, coomia-dip