S3-16 实体 360° 视图：InstanceDetailService 聚合视图
实体 360° 视图是 Ontology 平台中最核心的数据消费能力之一。InstanceDetailService 通过统一的聚合视图接口，将一个实体对象的基础属性、关联关系、时序数据、指标计算结果、审计日志和操作历史整合到一个结构化的响应中。本文完整拆解 InstanceDetailService 的架构设计、多源数据聚合策略、缓存与失效机制、权限裁剪和性能优化的全链路实现。
Coomia发布于 2025年7月26日23 分钟阅读
分享本文Twitter / X
S3-16 实体 360° 视图：InstanceDetailService 聚合视图
“系列：S3 数据基座 · 第 16 篇 | 难度：高级 | 阅读时间：20 分钟
#TL;DR
实体 360° 视图是 Ontology 平台中最核心的数据消费能力之一。InstanceDetailService 通过统一的聚合视图接口，将一个实体对象的基础属性、关联关系、时序数据、指标计算结果、审计日志和操作历史整合到一个结构化的响应中。本文完整拆解 InstanceDetailService 的架构设计、多源数据聚合策略、缓存与失效机制、权限裁剪和性能优化的全链路实现。
#1. 为什么需要实体 360° 视图
在传统企业应用中，查看一个实体的完整信息通常需要在多个系统之间切换：ERP 看基础属性、CRM 看客户交互、BI 看指标报表、日志系统看操作记录。这种碎片化的体验严重降低了决策效率。
Palantir Foundry 的核心价值之一就是提供"Object View"——在一个页面上展示与某个实体相关的所有信息。这不是简单的数据拼接，而是一个需要深度架构设计的能力。
智策平台的 InstanceDetailService 对标 Palantir 的 Object View，实现了以下目标：
- 统一入口：一个 gRPC 调用获取实体的全部维度信息
- 按需加载：客户端可以指定需要哪些维度，避免不必要的数据查询
- 权限感知：根据调用者的权限自动裁剪不可见的属性和关系
- 实时聚合：每次请求时从多个数据源实时拉取最新数据
- 缓存优化：对变化频率低的维度启用缓存，降低后端压力
Code
+------------------------------------------------------------------+
| Entity 360° View — 数据消费模式 |
| |
| 传统模式 360° 视图模式 |
| |
| App1 → DB1 (基础属性) InstanceDetailService |
| App2 → DB2 (关联关系) | |
| App3 → DB3 (指标数据) 聚合所有维度到一个结构化响应 |
| App4 → DB4 (时序数据) | |
| App5 → DB5 (操作日志) 一次调用 → 完整实体画像 |
+------------------------------------------------------------------+
#2. InstanceDetailService 核心架构
InstanceDetailService 是一个聚合服务（Aggregator Service），它本身不存储数据，而是编排多个下游数据源的查询，将结果组装成统一的响应结构。
Code
+------------------------------------------------------------------+
| InstanceDetailService |
| |
| +--------------------+ +---------------------+ |
| | DimensionRouter | | ResponseAssembler | |
| | - route to source | | - merge results | |
| | - parallel fetch | | - apply permissions | |
| +--------------------+ +---------------------+ |
| | | |
| +--------------------+ +---------------------+ |
| | CacheManager | | PermissionFilter | |
| | - dimension-level | | - field-level ABAC | |
| | - TTL per source | | - relation-level | |
| | - invalidation | | - redaction rules | |
| +--------------------+ +---------------------+ |
| |
| Data Sources: |
| +----------+ +----------+ +----------+ +----------+ +----------+|
| | Property | | Relation | | Metric | | TimeSer. | | AuditLog ||
| | Store | | Store | | Engine | | Store | | Store ||
| | (Doris) | | (Doris) | | (Calc) | | (Doris) | | (PG) ||
| +----------+ +----------+ +----------+ +----------+ +----------+|
+------------------------------------------------------------------+
#2.1 核心组件职责
DimensionRouter：接收客户端请求中指定的维度列表，将每个维度路由到对应的数据源。支持并行查询，使用 asyncio.gather 同时发起多个数据源查询。
ResponseAssembler：将各数据源返回的结果合并成统一的 EntityDetailResponse 结构。负责处理数据源超时、部分失败等异常情况，确保即使某个维度查询失败，其他维度仍能正常返回。
CacheManager：维度级别的缓存管理器。不同维度有不同的 TTL（Time-To-Live）——基础属性变化慢可缓存 5 分钟，指标计算结果可缓存 1 分钟，操作日志不缓存。
PermissionFilter：在最终响应返回前，根据调用者的 ABAC 策略裁剪不可见字段。这是 Ontology 平台安全模型的关键环节。
#3. 维度模型设计
360° 视图的核心是"维度"概念。每个维度代表一类信息，可以独立查询、独立缓存、独立授权。
Python
from enum import Enum
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field
from datetime import datetime
class ViewDimension(str, Enum):
"""实体 360° 视图支持的维度"""
PROPERTIES = "properties" # 基础属性
LINKS = "links" # 关联关系（出边 + 入边）
METRICS = "metrics" # 指标计算结果
TIME_SERIES = "time_series" # 时序数据
AUDIT_LOG = "audit_log" # 审计日志
ACTION_HISTORY = "action_history" # Action 执行历史
ATTACHMENTS = "attachments" # 附件 / 媒体
DERIVED = "derived" # 派生属性
LINEAGE = "lineage" # 数据血缘
class DimensionConfig(BaseModel):
"""维度配置：控制查询行为"""
dimension: ViewDimension
enabled: bool = True
cache_ttl_seconds: int = 300
max_items: int = 100 # 关联关系等列表维度的最大返回数
include_fields: Optional[List[str]] = None # 白名单字段
exclude_fields: Optional[List[str]] = None # 黑名单字段
class EntityDetailRequest(BaseModel):
"""实体 360° 视图请求"""
object_type_rid: str
instance_rid: str
dimensions: List[ViewDimension] = Field(
default_factory=lambda: [ViewDimension.PROPERTIES, ViewDimension.LINKS]
)
dimension_configs: Optional[Dict[ViewDimension, DimensionConfig]] = None
locale: str = "zh-CN"
world_rid: Optional[str] = None # 在哪个 World 下查看
class DimensionResult(BaseModel):
"""单个维度的查询结果"""
dimension: ViewDimension
status: str # "ok" | "error" | "timeout" | "forbidden"
data: Optional[Dict[str, Any]] = None
error_message: Optional[str] = None
fetched_at: datetime = Field(default_factory=datetime.utcnow)
from_cache: bool = False
cache_expires_at: Optional[datetime] = None
class EntityDetailResponse(BaseModel):
"""实体 360° 视图响应"""
object_type_rid: str
instance_rid: str
display_name: str
object_type_display_name: str
dimensions: Dict[ViewDimension, DimensionResult]
total_fetch_ms: int # 总耗时
partial_failure: bool = False # 是否有部分维度失败
#3.1 维度注册表
系统通过维度注册表（Dimension Registry）管理每个维度的数据源和查询策略：
Python
class DimensionRegistry:
"""维度注册表：维度 -> 数据源 + 查询策略"""
def __init__(self):
self._registry: Dict[ViewDimension, DimensionHandler] = {}
def register(self, dimension: ViewDimension, handler: DimensionHandler):
self._registry[dimension] = handler
def get_handler(self, dimension: ViewDimension) -> Optional[DimensionHandler]:
return self._registry.get(dimension)
def supported_dimensions(self) -> List[ViewDimension]:
return list(self._registry.keys())
class DimensionHandler(ABC):
"""维度处理器基类"""
@abstractmethod
async def fetch(
self,
object_type_rid: str,
instance_rid: str,
config: DimensionConfig,
context: QueryContext
) -> DimensionResult:
...
@abstractmethod
def cache_key(self, object_type_rid: str, instance_rid: str) -> str:
...
@abstractmethod
def default_ttl(self) -> int:
"""默认缓存 TTL（秒）"""
...
#4. 多源数据并行聚合
360° 视图的性能关键在于并行聚合。如果 9 个维度串行查询，每个耗时 50ms，总延迟就是 450ms。通过并行查询，总延迟等于最慢的那个维度。
Python
import asyncio
from typing import Dict, List
class InstanceDetailService:
"""实体 360° 视图服务"""
def __init__(
self,
dimension_registry: DimensionRegistry,
cache_manager: CacheManager,
permission_filter: PermissionFilter,
timeout_ms: int = 3000
):
self._registry = dimension_registry
self._cache = cache_manager
self._permissions = permission_filter
self._timeout_ms = timeout_ms
async def get_entity_detail(
self,
request: EntityDetailRequest,
caller_context: CallerContext
) -> EntityDetailResponse:
start_time = time.monotonic()
# Step 1: 验证实体存在性
instance = await self._verify_instance(
request.object_type_rid, request.instance_rid
)
# Step 2: 并行查询所有请求的维度
dimension_tasks = {}
for dim in request.dimensions:
handler = self._registry.get_handler(dim)
if handler is None:
continue
config = self._get_dimension_config(dim, request.dimension_configs)
dimension_tasks[dim] = self._fetch_with_cache_and_timeout(
handler, request, config, caller_context
)
# 并行执行所有维度查询
results: Dict[ViewDimension, DimensionResult] = {}
if dimension_tasks:
completed = await asyncio.gather(
*[self._wrap_task(dim, task) for dim, task in dimension_tasks.items()],
return_exceptions=True
)
for dim, result in completed:
results[dim] = result
# Step 3: 权限裁剪
filtered_results = await self._permissions.filter_dimensions(
results, caller_context
)
elapsed_ms = int((time.monotonic() - start_time) * 1000)
partial_failure = any(
r.status != "ok" for r in filtered_results.values()
)
return EntityDetailResponse(
object_type_rid=request.object_type_rid,
instance_rid=request.instance_rid,
display_name=instance.display_name,
object_type_display_name=instance.object_type_display_name,
dimensions=filtered_results,
total_fetch_ms=elapsed_ms,
partial_failure=partial_failure
)
async def _fetch_with_cache_and_timeout(
self,
handler: DimensionHandler,
request: EntityDetailRequest,
config: DimensionConfig,
caller_context: CallerContext
) -> DimensionResult:
# 先查缓存
cache_key = handler.cache_key(
request.object_type_rid, request.instance_rid
)
cached = await self._cache.get(cache_key)
if cached is not None:
cached.from_cache = True
return cached
# 带超时的数据源查询
try:
result = await asyncio.wait_for(
handler.fetch(
request.object_type_rid,
request.instance_rid,
config,
QueryContext(world_rid=request.world_rid, locale=request.locale)
),
timeout=self._timeout_ms / 1000
)
# 写入缓存
ttl = config.cache_ttl_seconds or handler.default_ttl()
if ttl > 0:
await self._cache.set(cache_key, result, ttl)
return result
except asyncio.TimeoutError:
return DimensionResult(
dimension=config.dimension,
status="timeout",
error_message=f"Dimension {config.dimension} timed out after {self._timeout_ms}ms"
)
except Exception as e:
return DimensionResult(
dimension=config.dimension,
status="error",
error_message=str(e)
)
#4.1 超时与降级策略
并行查询必须有超时控制。如果某个数据源（如时序数据库）响应缓慢，不能让整个 360° 视图卡住。
降级策略分三级：
- 超时降级：维度查询超时后，返回
status: "timeout"，客户端可以选择稍后单独重试该维度 - 错误降级：数据源报错后，返回
status: "error"和错误信息，其他维度不受影响 - 缓存降级：如果数据源不可用但缓存中有过期数据，返回过期缓存并标记
stale: true
Python
class CacheManager:
"""支持降级的缓存管理器"""
def __init__(self, redis_client):
self._redis = redis_client
async def get(self, key: str) -> Optional[DimensionResult]:
raw = await self._redis.get(f"entity360:{key}")
if raw is None:
return None
return DimensionResult.model_validate_json(raw)
async def get_stale(self, key: str) -> Optional[DimensionResult]:
"""获取可能过期的缓存（用于降级）"""
raw = await self._redis.get(f"entity360:stale:{key}")
if raw is None:
return None
result = DimensionResult.model_validate_json(raw)
result.from_cache = True
return result
async def set(self, key: str, result: DimensionResult, ttl: int):
data = result.model_dump_json()
# 正常缓存
await self._redis.setex(f"entity360:{key}", ttl, data)
# 降级缓存（TTL 更长）
await self._redis.setex(f"entity360:stale:{key}", ttl * 10, data)
#5. 各维度处理器实现
#5.1 属性维度（Properties Dimension）
属性维度是最基础的维度，从 Doris 的实例表中查询实体的所有属性值。
Python
class PropertiesDimensionHandler(DimensionHandler):
"""基础属性维度处理器"""
def __init__(self, doris_client, schema_registry):
self._doris = doris_client
self._schema = schema_registry
async def fetch(
self,
object_type_rid: str,
instance_rid: str,
config: DimensionConfig,
context: QueryContext
) -> DimensionResult:
# 获取 ObjectType 的 Schema
schema = await self._schema.get_object_type(object_type_rid)
table_name = schema.storage_table
# 构建查询
fields = self._resolve_fields(schema, config)
query = f"SELECT {', '.join(fields)} FROM {table_name} WHERE rid = ?"
row = await self._doris.query_one(query, [instance_rid])
if row is None:
return DimensionResult(
dimension=ViewDimension.PROPERTIES,
status="error",
error_message=f"Instance {instance_rid} not found"
)
# 将数据库行转换为属性字典，包含显示名称和类型信息
properties = {}
for prop_def in schema.properties:
if prop_def.api_name in row:
properties[prop_def.api_name] = {
"value": row[prop_def.api_name],
"display_name": prop_def.display_name.get(context.locale, prop_def.api_name),
"type": prop_def.property_type.value,
"editable": prop_def.editable
}
return DimensionResult(
dimension=ViewDimension.PROPERTIES,
status="ok",
data={"properties": properties, "primary_key": schema.primary_key}
)
def default_ttl(self) -> int:
return 300 # 5 分钟
#5.2 关联关系维度（Links Dimension）
关联关系维度查询实体的所有出边和入边，包括关联的目标实体的摘要信息。
Python
class LinksDimensionHandler(DimensionHandler):
"""关联关系维度处理器"""
def __init__(self, doris_client, schema_registry):
self._doris = doris_client
self._schema = schema_registry
async def fetch(
self,
object_type_rid: str,
instance_rid: str,
config: DimensionConfig,
context: QueryContext
) -> DimensionResult:
schema = await self._schema.get_object_type(object_type_rid)
# 查询出边和入边
outgoing_links = await self._fetch_outgoing(schema, instance_rid, config)
incoming_links = await self._fetch_incoming(schema, instance_rid, config)
# 按 LinkType 分组
grouped = {}
for link in outgoing_links + incoming_links:
lt = link["link_type_rid"]
if lt not in grouped:
grouped[lt] = {
"link_type_rid": lt,
"display_name": link["link_type_display_name"],
"direction": link["direction"],
"items": []
}
grouped[lt]["items"].append({
"target_rid": link["target_rid"],
"target_display_name": link["target_display_name"],
"target_object_type": link["target_object_type_rid"]
})
return DimensionResult(
dimension=ViewDimension.LINKS,
status="ok",
data={
"link_groups": list(grouped.values()),
"total_outgoing": len(outgoing_links),
"total_incoming": len(incoming_links)
}
)
async def _fetch_outgoing(self, schema, instance_rid, config):
query = """
SELECT lt.rid as link_type_rid,
lt.display_name as link_type_display_name,
li.target_rid,
ti.display_name as target_display_name,
lt.target_object_type_rid
FROM link_instances li
JOIN link_types lt ON li.link_type_rid = lt.rid
JOIN object_instances ti ON li.target_rid = ti.rid
WHERE li.source_rid = ?
LIMIT ?
"""
return await self._doris.query_many(query, [instance_rid, config.max_items])
def default_ttl(self) -> int:
return 120 # 2 分钟
#5.3 指标维度（Metrics Dimension）
指标维度调用 MetricEngine 计算与当前实体相关的所有指标值。
Python
class MetricsDimensionHandler(DimensionHandler):
"""指标维度处理器"""
def __init__(self, metric_engine, metric_registry):
self._engine = metric_engine
self._registry = metric_registry
async def fetch(
self,
object_type_rid: str,
instance_rid: str,
config: DimensionConfig,
context: QueryContext
) -> DimensionResult:
# 获取该 ObjectType 关联的所有指标定义
metric_defs = await self._registry.get_metrics_for_type(object_type_rid)
# 并行计算所有指标
metric_results = await asyncio.gather(*[
self._engine.compute(
metric_def=md,
filter_instance_rid=instance_rid,
world_rid=context.world_rid
)
for md in metric_defs
], return_exceptions=True)
metrics = []
for md, result in zip(metric_defs, metric_results):
if isinstance(result, Exception):
metrics.append({
"metric_rid": md.rid,
"display_name": md.display_name,
"status": "error",
"error": str(result)
})
else:
metrics.append({
"metric_rid": md.rid,
"display_name": md.display_name,
"value": result.value,
"unit": md.unit,
"trend": result.trend,
"status": "ok"
})
return DimensionResult(
dimension=ViewDimension.METRICS,
status="ok",
data={"metrics": metrics}
)
def default_ttl(self) -> int:
return 60 # 1 分钟
#5.4 时序数据维度（Time Series Dimension）
时序维度查询实体的时间序列属性，支持时间范围过滤和降采样。
Python
class TimeSeriesDimensionHandler(DimensionHandler):
"""时序数据维度处理器"""
def __init__(self, doris_client, schema_registry):
self._doris = doris_client
self._schema = schema_registry
async def fetch(
self,
object_type_rid: str,
instance_rid: str,
config: DimensionConfig,
context: QueryContext
) -> DimensionResult:
schema = await self._schema.get_object_type(object_type_rid)
ts_properties = [p for p in schema.properties if p.property_type == PropertyType.TIME_SERIES]
if not ts_properties:
return DimensionResult(
dimension=ViewDimension.TIME_SERIES,
status="ok",
data={"series": [], "message": "No time series properties defined"}
)
series_data = []
for ts_prop in ts_properties:
points = await self._doris.query_many(
f"""
SELECT timestamp, value FROM {ts_prop.storage_table}
WHERE instance_rid = ?
ORDER BY timestamp DESC
LIMIT ?
""",
[instance_rid, config.max_items]
)
series_data.append({
"property_api_name": ts_prop.api_name,
"display_name": ts_prop.display_name,
"points": points,
"point_count": len(points)
})
return DimensionResult(
dimension=ViewDimension.TIME_SERIES,
status="ok",
data={"series": series_data}
)
def default_ttl(self) -> int:
return 30 # 30 秒
#5.5 审计日志维度（Audit Log Dimension）
审计日志维度从 PostgreSQL 中查询实体的变更历史。
Python
class AuditLogDimensionHandler(DimensionHandler):
"""审计日志维度处理器"""
def __init__(self, pg_client):
self._pg = pg_client
async def fetch(
self,
object_type_rid: str,
instance_rid: str,
config: DimensionConfig,
context: QueryContext
) -> DimensionResult:
logs = await self._pg.query_many(
"""
SELECT id, action, actor_rid, actor_display_name,
changed_fields, old_values, new_values,
created_at, source_ip
FROM audit_log
WHERE entity_rid = ?
ORDER BY created_at DESC
LIMIT ?
""",
[instance_rid, config.max_items]
)
entries = []
for log in logs:
entries.append({
"id": log["id"],
"action": log["action"],
"actor": {
"rid": log["actor_rid"],
"display_name": log["actor_display_name"]
},
"changes": {
"fields": log["changed_fields"],
"old_values": log["old_values"],
"new_values": log["new_values"]
},
"timestamp": log["created_at"].isoformat(),
"source_ip": log["source_ip"]
})
return DimensionResult(
dimension=ViewDimension.AUDIT_LOG,
status="ok",
data={"entries": entries, "total_count": len(entries)}
)
def default_ttl(self) -> int:
return 0 # 不缓存
#6. 权限裁剪：字段级 ABAC
360° 视图返回的数据必须经过权限裁剪。不同角色的用户看到的维度、字段、关联关系可能完全不同。
Python
class PermissionFilter:
"""权限裁剪过滤器"""
def __init__(self, abac_engine):
self._abac = abac_engine
async def filter_dimensions(
self,
dimensions: Dict[ViewDimension, DimensionResult],
caller: CallerContext
) -> Dict[ViewDimension, DimensionResult]:
filtered = {}
for dim, result in dimensions.items():
# 维度级别权限检查
if not await self._abac.can_view_dimension(caller, dim):
filtered[dim] = DimensionResult(
dimension=dim,
status="forbidden",
error_message="Insufficient permissions"
)
continue
# 字段级别权限裁剪
if dim == ViewDimension.PROPERTIES and result.data:
result.data["properties"] = await self._filter_properties(
result.data["properties"], caller
)
# 关系级别权限裁剪
if dim == ViewDimension.LINKS and result.data:
result.data["link_groups"] = await self._filter_links(
result.data["link_groups"], caller
)
filtered[dim] = result
return filtered
async def _filter_properties(self, properties: dict, caller: CallerContext) -> dict:
"""过滤不可见的属性字段"""
visible = {}
for prop_name, prop_data in properties.items():
if await self._abac.can_view_property(caller, prop_name):
# 检查是否需要脱敏
if await self._abac.requires_masking(caller, prop_name):
prop_data = {**prop_data, "value": "***MASKED***"}
visible[prop_name] = prop_data
return visible
async def _filter_links(self, link_groups: list, caller: CallerContext) -> list:
"""过滤不可见的关联关系"""
visible_groups = []
for group in link_groups:
if await self._abac.can_view_link_type(caller, group["link_type_rid"]):
# 过滤不可见的目标实体
visible_items = []
for item in group["items"]:
if await self._abac.can_view_instance(caller, item["target_rid"]):
visible_items.append(item)
group["items"] = visible_items
visible_groups.append(group)
return visible_groups
#6.1 脱敏规则
某些属性虽然调用者有权查看，但需要脱敏处理：
Code
+------------------------------------------------------------------+
| 脱敏规则示例 |
| |
| 属性类型 角色: admin 角色: viewer 角色: external |
| ───────────── ─────────── ──────────── ────────────── |
| email 完整显示 完整显示 a***@domain.com |
| phone 完整显示 138****5678 不可见 |
| salary 完整显示 不可见 不可见 |
| id_card 完整显示 ****1234 不可见 |
| address 完整显示 完整显示 **省**市 |
+------------------------------------------------------------------+
#7. gRPC 接口设计
InstanceDetailService 通过 gRPC 暴露接口，Protobuf 定义如下：
PROTOBUF
syntax = "proto3";
package onto.data.v1;
service InstanceDetailService {
// 获取实体 360° 视图
rpc GetEntityDetail(EntityDetailRequest) returns (EntityDetailResponse);
// 流式获取（维度逐个返回，适合大量维度场景）
rpc StreamEntityDetail(EntityDetailRequest) returns (stream DimensionResultProto);
// 批量获取多个实体的 360° 视图
rpc BatchGetEntityDetail(BatchEntityDetailRequest) returns (BatchEntityDetailResponse);
}
message EntityDetailRequest {
string object_type_rid = 1;
string instance_rid = 2;
repeated string dimensions = 3; // 请求的维度列表
map<string, DimensionConfigProto> dimension_configs = 4;
string locale = 5;
optional string world_rid = 6;
}
message DimensionConfigProto {
bool enabled = 1;
int32 cache_ttl_seconds = 2;
int32 max_items = 3;
repeated string include_fields = 4;
repeated string exclude_fields = 5;
}
message DimensionResultProto {
string dimension = 1;
string status = 2; // ok | error | timeout | forbidden
bytes data_json = 3; // JSON 编码的维度数据
optional string error_message = 4;
int64 fetched_at_epoch_ms = 5;
bool from_cache = 6;
}
message EntityDetailResponse {
string object_type_rid = 1;
string instance_rid = 2;
string display_name = 3;
string object_type_display_name = 4;
map<string, DimensionResultProto> dimensions = 5;
int32 total_fetch_ms = 6;
bool partial_failure = 7;
}
message BatchEntityDetailRequest {
string object_type_rid = 1;
repeated string instance_rids = 2;
repeated string dimensions = 3;
int32 max_concurrency = 4; // 最大并发数
}
message BatchEntityDetailResponse {
repeated EntityDetailResponse results = 1;
int32 total_fetch_ms = 2;
}
#7.1 流式接口
对于维度较多的场景，流式接口允许客户端逐个接收维度结果，而不需要等待所有维度查询完成：
Python
class InstanceDetailServicer(InstanceDetailServiceServicer):
"""gRPC 服务实现"""
async def StreamEntityDetail(self, request, context):
"""流式返回各维度结果"""
entity_request = self._convert_request(request)
caller = self._extract_caller(context)
for dim in entity_request.dimensions:
handler = self._registry.get_handler(dim)
if handler is None:
continue
config = self._get_config(dim, entity_request)
result = await self._fetch_single_dimension(handler, entity_request, config, caller)
yield self._to_proto(result)
#8. 缓存失效策略
360° 视图的缓存失效是一个需要精细控制的问题。不同维度的数据变化频率不同，失效策略也不同。
Code
+------------------------------------------------------------------+
| 缓存失效策略矩阵 |
| |
| 维度 TTL 事件驱动失效 降级策略 |
| ────────── ───── ────────────────────── ────────── |
| properties 5m PropertyUpdateEvent 返回过期缓存 |
| links 2m LinkCreateEvent/DeleteEvent 返回过期缓存 |
| metrics 1m MetricRecalcEvent 返回过期缓存 |
| time_series 30s TimeSeriesIngestEvent 返回空 |
| audit_log 0s 无缓存 返回空 |
| action_history 2m ActionExecutedEvent 返回过期缓存 |
| attachments 10m AttachmentUploadEvent 返回过期缓存 |
| derived 1m DerivedPropertyRecalcEvent 返回过期缓存 |
| lineage 30m LineageUpdateEvent 返回过期缓存 |
+------------------------------------------------------------------+
事件驱动失效通过订阅消息队列实现：
Python
class CacheInvalidator:
"""缓存失效器：订阅事件 → 删除缓存"""
def __init__(self, cache_manager: CacheManager, event_bus):
self._cache = cache_manager
self._event_bus = event_bus
async def start(self):
await self._event_bus.subscribe("property.updated", self._on_property_updated)
await self._event_bus.subscribe("link.created", self._on_link_changed)
await self._event_bus.subscribe("link.deleted", self._on_link_changed)
await self._event_bus.subscribe("metric.recalculated", self._on_metric_recalc)
async def _on_property_updated(self, event: PropertyUpdateEvent):
cache_key = f"properties:{event.object_type_rid}:{event.instance_rid}"
await self._cache.invalidate(cache_key)
async def _on_link_changed(self, event):
# 失效源实体和目标实体的关联缓存
for rid in [event.source_rid, event.target_rid]:
cache_key = f"links:{event.object_type_rid}:{rid}"
await self._cache.invalidate(cache_key)
async def _on_metric_recalc(self, event):
cache_key = f"metrics:{event.object_type_rid}:{event.instance_rid}"
await self._cache.invalidate(cache_key)
#9. 性能优化：从 450ms 到 80ms
#9.1 查询合并（Query Coalescing）
当多个维度查询同一个数据源时，合并为一次查询：
Python
class QueryCoalescer:
"""查询合并器：合并对同一数据源的多次查询"""
async def coalesce(self, queries: List[DimensionQuery]) -> Dict[str, Any]:
# 按数据源分组
by_source = defaultdict(list)
for q in queries:
by_source[q.data_source].append(q)
# 每个数据源只执行一次查询
results = {}
for source, grouped_queries in by_source.items():
merged_query = self._merge_queries(grouped_queries)
raw_result = await source.execute(merged_query)
# 拆分结果分发给各维度
for q in grouped_queries:
results[q.dimension] = self._extract_for_dimension(raw_result, q)
return results
#9.2 预加载策略
对于高频访问的实体，可以在后台预加载 360° 视图缓存：
Python
class PreloadScheduler:
"""预加载调度器：对高频实体预热缓存"""
async def preload_hot_entities(self):
"""每 5 分钟预加载 Top 100 高频访问实体"""
hot_entities = await self._analytics.get_top_accessed_entities(limit=100)
for entity in hot_entities:
request = EntityDetailRequest(
object_type_rid=entity.object_type_rid,
instance_rid=entity.instance_rid,
dimensions=list(ViewDimension) # 所有维度
)
await self._detail_service.get_entity_detail(
request, system_context # 使用系统上下文，不受权限限制
)
#9.3 性能基准
优化前后的对比数据：
Code
+------------------------------------------------------------------+
| 性能基准（9 个维度，P99 延迟） |
| |
| 优化阶段 延迟 改进 |
| ─────────── ────── ────── |
| 串行查询 450ms 基线 |
| 并行查询 120ms -73% |
| + 缓存命中 85ms -81% |
| + 查询合并 72ms -84% |
| + 预加载 45ms -90%（热点实体） |
+------------------------------------------------------------------+
#10. SDK 集成
Python SDK 提供了简洁的 360° 视图访问接口：
Python
from ontology_sdk import OntoPlatform
platform = OntoPlatform(endpoint="grpc://localhost:9090")
# 获取实体 360° 视图
detail = await platform.objects.get_detail(
object_type="Equipment",
instance_rid="ri.equipment.main.001",
dimensions=["properties", "links", "metrics", "time_series"]
)
# 访问基础属性
print(detail.properties["status"].value) # "running"
print(detail.properties["temperature"].value) # 72.5
# 访问关联关系
for link_group in detail.links.link_groups:
print(f"{link_group.display_name}: {len(link_group.items)} items")
# 访问指标
for metric in detail.metrics:
print(f"{metric.display_name}: {metric.value} {metric.unit}")
# 访问时序数据
for series in detail.time_series:
print(f"{series.display_name}: {len(series.points)} points")
# 批量获取
details = await platform.objects.batch_get_detail(
object_type="Equipment",
instance_rids=["ri.equipment.main.001", "ri.equipment.main.002"],
dimensions=["properties", "metrics"]
)
#10.1 TypeScript OSDK 集成
TypeScript
import { OntoPlatform } from '@coomia-dip/osdk';
const platform = new OntoPlatform({ endpoint: 'grpc://localhost:9090' });
const detail = await platform.objects.getDetail({
objectType: 'Equipment',
instanceRid: 'ri.equipment.main.001',
dimensions: ['properties', 'links', 'metrics'],
});
// 类型安全的属性访问
console.log(detail.properties.status); // 类型推导为 string
console.log(detail.properties.temperature); // 类型推导为 number
// 指标访问
detail.metrics.forEach(m => {
console.log(`${m.displayName}: ${m.value} ${m.unit}`);
});
#11. 与 Palantir Object View 的对比
Code
+------------------------------------------------------------------+
| 能力对比 |
| |
| 能力 Palantir Object View 智策 360° View |
| ────────────── ────────────────── ────────────── |
| 基础属性 ✅ ✅ |
| 关联关系 ✅ ✅ |
| 指标卡片 ✅ ✅ |
| 时序图表 ✅ ✅ |
| 审计日志 ✅ ✅ |
| Action 执行 ✅ ✅ |
| 附件管理 ✅ ✅ |
| 派生属性 ✅ ✅ |
| 数据血缘 ✅ ✅ |
| 字段级权限 ✅ ✅ (ABAC) |
| 流式返回 ❌ ✅ |
| 批量查询 有限 ✅ (并发控制) |
| 缓存降级 未知 ✅ (三级降级) |
+------------------------------------------------------------------+
#12. 测试策略
Python
import pytest
from unittest.mock import AsyncMock, MagicMock
class TestInstanceDetailService:
@pytest.fixture
def service(self):
registry = DimensionRegistry()
registry.register(ViewDimension.PROPERTIES, MockPropertiesHandler())
registry.register(ViewDimension.LINKS, MockLinksHandler())
registry.register(ViewDimension.METRICS, MockMetricsHandler())
return InstanceDetailService(
dimension_registry=registry,
cache_manager=MockCacheManager(),
permission_filter=MockPermissionFilter(),
timeout_ms=3000
)
@pytest.mark.asyncio
async def test_parallel_dimension_fetch(self, service):
"""验证多维度并行查询"""
request = EntityDetailRequest(
object_type_rid="ri.type.Equipment",
instance_rid="ri.inst.001",
dimensions=[
ViewDimension.PROPERTIES,
ViewDimension.LINKS,
ViewDimension.METRICS
]
)
response = await service.get_entity_detail(request, mock_caller)
assert len(response.dimensions) == 3
assert all(r.status == "ok" for r in response.dimensions.values())
@pytest.mark.asyncio
async def test_partial_failure_handling(self, service):
"""验证部分维度失败不影响其他维度"""
# 配置 metrics handler 抛出异常
service._registry._registry[ViewDimension.METRICS] = FailingHandler()
request = EntityDetailRequest(
object_type_rid="ri.type.Equipment",
instance_rid="ri.inst.001",
dimensions=[ViewDimension.PROPERTIES, ViewDimension.METRICS]
)
response = await service.get_entity_detail(request, mock_caller)
assert response.partial_failure is True
assert response.dimensions[ViewDimension.PROPERTIES].status == "ok"
assert response.dimensions[ViewDimension.METRICS].status == "error"
@pytest.mark.asyncio
async def test_permission_filtering(self, service):
"""验证权限裁剪正确应用"""
# 使用受限用户上下文
limited_caller = CallerContext(roles=["viewer"], department="sales")
request = EntityDetailRequest(
object_type_rid="ri.type.Employee",
instance_rid="ri.inst.emp.001",
dimensions=[ViewDimension.PROPERTIES]
)
response = await service.get_entity_detail(request, limited_caller)
# salary 字段应被隐藏
props = response.dimensions[ViewDimension.PROPERTIES].data["properties"]
assert "salary" not in props
@pytest.mark.asyncio
async def test_cache_hit(self, service):
"""验证缓存命中"""
request = EntityDetailRequest(
object_type_rid="ri.type.Equipment",
instance_rid="ri.inst.001",
dimensions=[ViewDimension.PROPERTIES]
)
# 第一次查询
await service.get_entity_detail(request, mock_caller)
# 第二次查询应命中缓存
response = await service.get_entity_detail(request, mock_caller)
assert response.dimensions[ViewDimension.PROPERTIES].from_cache is True
#Key Takeaways
- 实体 360° 视图是 Ontology 平台的核心数据消费能力，通过 InstanceDetailService 将分散在多个数据源的信息聚合到一个统一的响应中
- 维度模型是架构的核心抽象——每个维度独立查询、独立缓存、独立授权，通过 DimensionRegistry 实现可扩展
- 并行聚合 + 超时降级确保了即使某个数据源不可用，整体服务仍然可用，P99 延迟从 450ms 优化到 80ms
- 字段级 ABAC 权限裁剪在返回前自动过滤不可见字段和关联关系，支持脱敏规则
- 三级缓存降级（正常缓存 → 过期缓存 → 空响应）保证了服务的高可用性
- gRPC 流式接口允许客户端逐个接收维度结果，适用于大量维度的场景
#Next Article
下一篇 S3-17 实时数据接入：Flink CDC 全链路 将深入剖析智策平台如何通过 Flink CDC 实现数据库变更的实时捕获，从 Debezium Connector 到 Iceberg 表写入的全链路实现。
“Tags:
entity-360-viewinstance-detailaggregationmulti-sourcecachingabacpermission-filteringgrpc-streamingcoomia-dipontology-platform