Palantir 的数据血缘追踪：每一个数值从哪里来？
想象你在超市买了一瓶橄榄油。食品安全法规要求你能追溯这瓶油的完整来历：
Coomia发布于 2025年6月15日22 分钟阅读
分享本文Twitter / X
“系列：S1 Palantir 解密 · 第 14 篇 | 难度：入门 | 阅读时间：15 分钟
Palantir 的数据血缘追踪：每一个数值从哪里来？
#TL;DR
- 数据血缘（Data Lineage）是 Foundry 的核心基础设施之一，它让用户能追踪任何数据点从原始来源经过所有转换步骤到最终展示的完整路径，支持实体级和字段级两种粒度。
- 血缘追踪不仅是审计合规的要求，更是建立数据信任的基石 -- 当 CEO 看到一个数字时，他可以点击追溯到原始数据源，理解每一步转换逻辑，判断数据是否可信。
- coomia-dip 通过 LineageService 实现了 11 个 RPC 接口，支持实体级和字段级血缘追踪、影响分析、工作流血缘，在开源生态中对标 Foundry 的血缘能力。
#1. 什么是数据血缘？
#1.1 一个直觉的类比
想象你在超市买了一瓶橄榄油。食品安全法规要求你能追溯这瓶油的完整来历：
Code
数据血缘的食品安全类比
================================================
橄榄油追溯:
橄榄树 (西班牙安达卢西亚, 地块 #A17)
→ 采摘 (2024-10-15, 工人组 #3)
→ 压榨 (冷榨, 机器 #M7, 温度 27°C)
→ 过滤 (三级过滤, 批次 #B2024-1015)
→ 灌装 (500ml, 产线 #L2, 批次 #F-88721)
→ 质检 (酸度 0.3%, 合格)
→ 物流 (冷链运输, 车辆 #T-9981)
→ 超市货架 (你看到的那一瓶)
数据血缘:
原始数据 (ERP系统, 表 sales_orders)
→ 清洗 (去重, 处理空值, Pipeline #P-001)
→ 转换 (汇率换算, 单位标准化, Transform #T-023)
→ 聚合 (按区域汇总, 月度累计, Transform #T-024)
→ 关联 (与客户维度表 JOIN, Transform #T-025)
→ 建模 (计算客户 LTV, Model #M-012)
→ 展示 (CEO 仪表盘上的 "本月营收: $12.5M")
数据血缘就是数据世界的"可追溯性" -- 你看到的每一个数值，都应该能追溯到它的源头。
#1.2 两种粒度的血缘
Code
实体级血缘 (Entity-Level / Dataset-Level)
================================================
追踪的是"数据集之间的关系":
Dataset A ─────→ Transform 1 ─────→ Dataset C
│
Dataset B ─────→ Transform 2 ──────────┘
问题: "Dataset C 依赖哪些上游数据集？"
答案: "Dataset A 和 Dataset B"
粒度: 粗 | 成本: 低 | 价值: 中等
字段级血缘 (Field-Level / Column-Level)
================================================
追踪的是"字段之间的关系":
Dataset A Dataset C
┌──────────────┐ ┌──────────────┐
│ order_id │────────────→│ order_id │
│ raw_amount │──┐ │ │
│ currency │──┼─Transform→│ amount_usd │
└──────────────┘ │ │ │
│ │ │
Dataset B │ │ │
┌──────────────┐ │ │ │
│ exchange_rate │──┘ │ │
│ customer_name│────────────→│ customer_name│
└──────────────┘ └──────────────┘
问题: "amount_usd 是怎么计算出来的？"
答案: "raw_amount × exchange_rate, 其中 raw_amount 来自
Dataset A, exchange_rate 来自 Dataset B"
粒度: 细 | 成本: 高 | 价值: 极高
#2. 为什么数据血缘至关重要？
#2.1 合规与审计
Code
合规场景
================================================
场景: 银行监管审查
监管官: "你们报给我的不良贷款率 2.3%，这个数字怎么来的？"
没有血缘:
分析师: "呃...让我查查...可能是从 A 系统取的数据..."
(3 天后) "我们确认是从 A 系统取的，但中间经过了 B 部门
的手工处理，具体逻辑我们需要再确认..."
(1 周后) "抱歉，负责这个流程的同事离职了..."
结果: 监管处罚、声誉损失
有血缘:
分析师: (点击数字) "这是完整的计算链路：
1. 源数据: 核心银行系统 loan_master 表
2. 清洗: 去除测试账户 (Pipeline #P-1234, 上次运行 2024-03-15)
3. 分类: 根据逾期天数分类 (>90天=不良, Transform #T-5678)
4. 聚合: 不良贷款总额 / 贷款总额 (Transform #T-5679)
5. 结果: 2.3% (更新时间: 2024-03-20 08:00)"
结果: 30 分钟搞定审查
#2.2 调试与问题定位
Code
调试场景
================================================
问题: "为什么本月销售报表的数字突然下降了 30%？"
没有血缘:
需要人工检查每一步:
1. 报表取数逻辑 → 看起来没问题
2. 数据仓库的表 → 数据量看起来对
3. ETL 任务 → 哪个 ETL？有 47 个...
4. 源系统 → 哪个源系统？有 12 个...
耗时: 2-3 天
有血缘:
1. 点击报表数字 → 看到依赖链
2. 发现 Transform #T-456 的输入数据集
上次更新时间是 3 天前 (之前是每天更新)
3. 追溯 Transform #T-456 的上游 →
发现源系统 CRM 的数据同步任务失败了
4. 原因: CRM 系统升级改了 API 格式
耗时: 15 分钟
#2.3 影响分析
Code
影响分析场景
================================================
问题: "如果我修改 customer_address 字段的格式，会影响什么？"
没有血缘:
"我不知道，让我问问每个团队..."
(可能遗漏，导致下游系统崩溃)
有血缘:
┌─────────────────────┐
│ customer_address │
│ (源: CRM系统) │
└──────────┬──────────┘
│
┌────────┼────────────────────┐
│ │ │
▼ ▼ ▼
物流系统 客户画像系统 合规报表
├─地址解析 ├─客户360视图 ├─反洗钱报告
├─配送路线 ├─区域分析 └─客户尽调报告
└─运费计算 └─邮件营销
影响范围: 3 个系统, 7 个下游转换, 12 个报表
风险等级: 高 (包含合规报表)
建议: 通知相关团队，安排兼容性变更窗口
#3. Foundry 如何追踪数据血缘
#3.1 Pipeline 级血缘自动采集
Foundry 的 Transforms 系统自动捕获代码执行过程中的数据流关系：
Python
# Foundry Transform 示例 -- 血缘自动捕获
from transforms.api import Transform, Input, Output
@transform(
order_input=Input("/datasets/raw/sales_orders"),
rate_input=Input("/datasets/ref/exchange_rates"),
output=Output("/datasets/clean/orders_usd")
)
def compute_orders_usd(order_input, rate_input, output):
"""
Foundry 自动捕获的血缘信息:
输入数据集:
- /datasets/raw/sales_orders
- /datasets/ref/exchange_rates
输出数据集:
- /datasets/clean/orders_usd
Transform 代码:
- 存储在版本控制中
- 每次执行记录代码版本
"""
orders = order_input.dataframe()
rates = rate_input.dataframe()
# 这个 JOIN 操作自动被记录为字段级血缘
result = orders.join(rates, on="currency")
# 这个计算自动被记录:
# amount_usd = amount * rate
result = result.withColumn(
"amount_usd",
result["amount"] * result["rate"]
)
output.write_dataframe(result)
# Foundry 自动记录的元数据:
# {
# "transform_rid": "ri.foundry.main.transform.abc123",
# "inputs": ["ri.foundry.main.dataset.orders", "ri.foundry.main.dataset.rates"],
# "output": "ri.foundry.main.dataset.orders_usd",
# "code_version": "git:abc123def",
# "execution_time": "2024-03-20T08:00:00Z",
# "field_lineage": {
# "order_id": {"source": "orders.order_id", "transform": "passthrough"},
# "amount_usd": {"source": ["orders.amount", "rates.rate"], "transform": "multiply"}
# }
# }
#3.2 Ontology 级血缘
血缘不仅追踪数据集之间的关系，还深入到 Ontology 对象层面：
Code
Ontology 血缘追踪
================================================
原始数据层:
┌─────────┐ ┌──────────┐ ┌─────────┐
│ CRM 系统 │ │ ERP 系统 │ │ IoT 平台 │
└────┬─────┘ └─────┬─────┘ └────┬─────┘
│ │ │
▼ ▼ ▼
清洗/转换层:
┌─────────┐ ┌──────────┐ ┌─────────┐
│客户数据 │ │订单数据 │ │设备数据 │
│(清洗后) │ │(清洗后) │ │(清洗后) │
└────┬─────┘ └─────┬─────┘ └────┬─────┘
│ │ │
▼ ▼ ▼
Ontology 对象层:
┌─────────────────────────────────────────┐
│ │
│ Customer 对象 Order 对象 │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ name │ │ order_id │ │
│ │ address │ │ amount │ │
│ │ lifetime_val │ │ status │ │
│ │ risk_score │ │ customer_ref │ │
│ └──────────────┘ └──────────────┘ │
│ │ │ │
│ └───── places ──────┘ │
│ │
│ lifetime_val 的血缘: │
│ = SUM(Order.amount) │
│ WHERE Order.customer_ref = this │
│ 来源: ERP.sales_orders.total_amount │
│ 转换: 汇率换算 → 月度聚合 → 累计 │
│ │
└─────────────────────────────────────────┘
#3.3 血缘可视化
Code
Foundry 血缘可视化 UI (概念示意)
================================================
┌──────────────────────────────────────────────────────┐
│ Lineage Explorer [X] │
│ │
│ 当前节点: "Monthly Revenue Report" │
│ ┌────────────────────────────────────────────────┐ │
│ │ │ │
│ │ [上游] │ │
│ │ │ │
│ │ CRM ──→ Clean ──→ Customer ──┐ │ │
│ │ ├──→ Revenue ──→│ │
│ │ ERP ──→ Clean ──→ Orders ────┘ Calc │ │
│ │ │ │ │
│ │ Rates ─────────────────────────────┘ │ │
│ │ │ │
│ │ [下游] │ │
│ │ │ │
│ │ ──→ Executive Dashboard │ │
│ │ ──→ Board Report │ │
│ │ ──→ SEC Filing (10-K) │ │
│ │ │ │
│ └────────────────────────────────────────────────┘ │
│ │
│ 深度: [1] [2] [3] [全部] 方向: [上游] [下游] [双向] │
│ │
│ ┌─ 节点详情 ───────────────────────────────────────┐ │
│ │ Revenue Calc (Transform #T-789) │ │
│ │ 最后执行: 2024-03-20 08:00 UTC │ │
│ │ 代码版本: main@abc123d │ │
│ │ 执行耗时: 4m 32s │ │
│ │ 输出行数: 1,247,831 │ │
│ │ 所有者: data-engineering@company.com │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────┘
#4. 字段级血缘的技术实现
#4.1 静态分析 vs 运行时捕获
实现字段级血缘有两种主要方法：
Code
方法对比
================================================
静态分析（解析代码/SQL）:
SQL: SELECT a.id, a.amount * b.rate AS amount_usd
FROM orders a JOIN rates b ON a.currency = b.code
解析结果:
amount_usd ← orders.amount, rates.rate (乘法运算)
id ← orders.id (直通)
优点: 不需要执行代码, 提前获得血缘
缺点: 复杂逻辑难以解析 (UDF, 动态 SQL)
运行时捕获（执行时记录）:
在 DataFrame 引擎中植入追踪:
每一列操作都记录来源
result["amount_usd"] = df["amount"] * df["rate"]
→ 记录: amount_usd = multiply(orders.amount, rates.rate)
优点: 精确, 支持复杂逻辑
缺点: 有运行时开销, 需要实际执行
Foundry 的方法: 两者结合
1. 静态分析获取初步血缘（快速）
2. 运行时验证和补充（精确）
3. 缓存结果避免重复计算
#4.2 血缘存储模型
Code
血缘数据模型
================================================
// 节点（数据实体）
Node {
id: string // 唯一标识
type: enum // DATASET | FIELD | TRANSFORM | MODEL
name: string // 人类可读名称
namespace: string // 所属项目/空间
metadata: map<string, any> // 扩展属性
}
// 边（血缘关系）
Edge {
source_id: string // 上游节点
target_id: string // 下游节点
type: enum // DERIVES_FROM | TRANSFORMS_TO | COMPUTES
transform: string // 转换描述 (e.g., "SUM", "JOIN", "FILTER")
confidence: float // 置信度 (静态分析可能不精确)
created_at: timestamp // 创建时间
created_by: string // 谁创建的 (Pipeline ID)
}
// 图存储选择
┌────────────────────────────────────────────────┐
│ 存储方案选型 │
│ │
│ Neo4j / JanusGraph: │
│ ✅ 图遍历性能好 │
│ ✅ 路径查询原生支持 │
│ ❌ 运维复杂 │
│ │
│ PostgreSQL + 递归 CTE: │
│ ✅ 运维简单 │
│ ✅ 事务支持好 │
│ ❌ 深度遍历性能有限 │
│ │
│ Foundry 选择: 自研图存储 │
│ coomia-dip 选择: PostgreSQL + 缓存优化 │
└────────────────────────────────────────────────┘
#5. 影响分析：改一个字段，谁会受影响？
#5.1 上游追溯
Code
上游追溯 (Upstream Tracing)
================================================
问题: "CEO 仪表盘上的 '季度营收' 数字来自哪里？"
查询: trace_upstream("dashboard.quarterly_revenue")
结果 (从下往上):
Layer 4 (展示): CEO Dashboard → quarterly_revenue
↑
Layer 3 (聚合): Revenue Aggregation Transform
= SUM(monthly_revenue) WHERE quarter = Q1
↑
Layer 2 (计算): Monthly Revenue Transform
= SUM(order_amount_usd) GROUP BY month
↑
Layer 1 (清洗): Order Cleaning Pipeline
= orders WHERE status != 'cancelled'
AND test_account = false
↑
Layer 0 (源): ERP System → sales_orders table
Source: SAP S/4HANA
Update frequency: Real-time CDC
Last sync: 2024-03-20 07:55:00
#5.2 下游影响
Code
下游影响分析 (Downstream Impact)
================================================
问题: "如果修改 exchange_rates 表的更新频率，会影响什么？"
查询: trace_downstream("ref.exchange_rates")
结果 (从上往下):
ref.exchange_rates
│
├──→ orders_usd (Transform #T-023)
│ ├──→ monthly_revenue (Transform #T-024)
│ │ ├──→ CEO Dashboard (Report)
│ │ ├──→ Board Report (Report)
│ │ └──→ SEC Filing (Report) ⚠️ 合规关键
│ │
│ └──→ customer_ltv (Model #M-012)
│ ├──→ Customer Risk Score (Model #M-015)
│ └──→ Marketing Campaign Targeting (App)
│
└──→ fx_pnl_report (Transform #T-067)
└──→ Treasury Dashboard (Report)
影响统计:
- 直接下游: 2 个 Transform
- 间接下游: 7 个数据集, 4 个报表, 1 个 ML 模型
- 合规影响: SEC Filing (高优先级)
- 建议: 变更前通知财务团队和合规团队
#6. 与主流数据血缘工具对比
| 维度 | Foundry Lineage | Apache Atlas | OpenLineage | DataHub |
|---|---|---|---|---|
| 血缘粒度 | 实体 + 字段级 | 主要实体级 | 实体 + 字段级 | 实体 + 字段级 |
| 自动采集 | 原生集成 | 需要 Hook | 需要集成 SDK | 需要 Ingestion |
| 实时性 | 执行时即刻更新 | 近实时 | 事件驱动 | 可配置 |
| Ontology 集成 | 深度集成 | 无 | 无 | 有限 |
| 影响分析 | 内置 UI | 有限 | 需自建 | 内置 |
| 版本化血缘 | 支持 (与代码版本关联) | 不支持 | 不支持 | 有限 |
| 权限集成 | 与 ACL 完全集成 | Ranger 集成 | 不涉及 | 有限 |
| 可视化 | 内置高质量 UI | 基础 UI | 无 UI | 内置 UI |
| 开源 | 否 | 是 | 是 | 是 |
| 适用规模 | 企业级 | 大规模 | 灵活 | 大规模 |
#各工具的定位差异
Code
数据血缘工具生态
================================================
自动化程度
▲
│
Foundry ────┤ ← 完全自动, 与平台深度集成
│
DataHub ─────┤ ← 需要配置 Ingestion Source
│
OpenLineage ────┤ ← 需要在代码中集成 SDK
│
Apache Atlas ────┤ ← 需要 Hook + 手动配置
│
手动维护 ─────┤ ← Excel 或 Wiki 记录 (不推荐)
│
└──────────────────────────→ 灵活性
Foundry: "全自动但锁定在平台内"
DataHub: "半自动, 支持多源, 社区活跃"
OpenLineage: "标准化协议, 适合异构环境"
Atlas: "Hadoop 生态原生, 大数据场景"
#7. coomia-dip 的血缘实现
#7.1 LineageService 架构
coomia-dip 的 LineageService 提供了 11 个 RPC 接口，覆盖血缘管理的完整生命周期：
Code
coomia-dip LineageService RPC 接口
================================================
┌───────────────────────────────────────────────┐
│ LineageService (gRPC) │
│ │
│ 基础 CRUD: │
│ 1. RecordLineage - 记录血缘关系 │
│ 2. GetLineage - 查询特定实体血缘 │
│ 3. DeleteLineage - 删除血缘记录 │
│ │
│ 图遍历: │
│ 4. TraceUpstream - 上游追溯 │
│ 5. TraceDownstream - 下游追踪 │
│ 6. GetFullGraph - 获取完整血缘图 │
│ │
│ 分析: │
│ 7. AnalyzeImpact - 影响分析 │
│ 8. FindRootCause - 根因定位 │
│ │
│ 字段级: │
│ 9. RecordFieldLineage - 记录字段级血缘 │
│ 10. GetFieldLineage - 查询字段级血缘 │
│ │
│ 工作流: │
│ 11. GetWorkflowLineage - 获取工作流血缘 │
│ │
└───────────────────────────────────────────────┘
#7.2 数据模型
PROTOBUF
// coomia-dip 血缘 Protobuf 定义 (简化版)
syntax = "proto3";
package onto.lineage.v1;
// 血缘节点
message LineageNode {
string id = 1;
string name = 2;
NodeType type = 3;
string namespace = 4;
map<string, string> properties = 5;
google.protobuf.Timestamp created_at = 6;
}
enum NodeType {
NODE_TYPE_UNSPECIFIED = 0;
DATASET = 1;
FIELD = 2;
TRANSFORM = 3;
MODEL = 4;
ONTOLOGY_TYPE = 5;
ONTOLOGY_PROPERTY = 6;
WORKFLOW = 7;
}
// 血缘边
message LineageEdge {
string source_id = 1;
string target_id = 2;
EdgeType type = 3;
string transform_description = 4;
float confidence = 5;
google.protobuf.Timestamp recorded_at = 6;
string recorded_by = 7; // Pipeline or User ID
}
enum EdgeType {
EDGE_TYPE_UNSPECIFIED = 0;
DERIVES_FROM = 1; // 数据派生
TRANSFORMS_TO = 2; // 转换关系
COMPUTES = 3; // 计算关系
REFERENCES = 4; // 引用关系
AGGREGATES = 5; // 聚合关系
}
// 影响分析结果
message ImpactAnalysisResult {
string source_id = 1;
repeated ImpactedEntity impacted_entities = 2;
int32 total_impacted = 3;
RiskLevel risk_level = 4;
}
message ImpactedEntity {
LineageNode node = 1;
int32 distance = 2; // 与源的距离
repeated string path = 3; // 影响路径
string impact_type = 4; // 影响类型描述
}
enum RiskLevel {
RISK_UNSPECIFIED = 0;
LOW = 1;
MEDIUM = 2;
HIGH = 3;
CRITICAL = 4;
}
#7.3 实现示例
Python
# coomia-dip LineageService 核心实现 (简化版)
from typing import List, Optional, Set
from dataclasses import dataclass, field
@dataclass
class LineageGraph:
"""内存中的血缘图结构"""
nodes: dict[str, LineageNode] = field(default_factory=dict)
forward_edges: dict[str, list[LineageEdge]] = field(default_factory=dict)
backward_edges: dict[str, list[LineageEdge]] = field(default_factory=dict)
class LineageServiceImpl:
"""血缘服务实现"""
def __init__(self, db_session, cache_client):
self.db = db_session
self.cache = cache_client
async def trace_upstream(
self,
entity_id: str,
max_depth: int = 10,
include_field_lineage: bool = False
) -> LineageGraph:
"""
上游追溯: 从目标实体回溯所有上游依赖
算法: BFS 图遍历
优化: Redis 缓存热点路径
"""
visited: Set[str] = set()
queue = [(entity_id, 0)]
graph = LineageGraph()
while queue:
current_id, depth = queue.pop(0)
if current_id in visited or depth > max_depth:
continue
visited.add(current_id)
# 获取节点信息
node = await self._get_node(current_id)
if node:
graph.nodes[current_id] = node
# 获取上游边
upstream_edges = await self._get_upstream_edges(current_id)
for edge in upstream_edges:
graph.backward_edges.setdefault(current_id, []).append(edge)
queue.append((edge.source_id, depth + 1))
if include_field_lineage:
field_edges = await self._get_field_lineage(
edge.source_id, current_id
)
for fe in field_edges:
graph.backward_edges.setdefault(
fe.target_id, []
).append(fe)
return graph
async def analyze_impact(
self,
entity_id: str,
change_type: str = "SCHEMA_CHANGE"
) -> ImpactAnalysisResult:
"""
影响分析: 评估修改某实体后的下游影响
返回: 受影响的实体列表 + 风险等级
"""
downstream_graph = await self.trace_downstream(
entity_id, max_depth=20
)
impacted = []
max_risk = RiskLevel.LOW
for node_id, node in downstream_graph.nodes.items():
if node_id == entity_id:
continue
distance = self._calculate_distance(
downstream_graph, entity_id, node_id
)
path = self._find_path(
downstream_graph, entity_id, node_id
)
# 评估风险
risk = self._assess_risk(node, change_type)
if risk.value > max_risk.value:
max_risk = risk
impacted.append(ImpactedEntity(
node=node,
distance=distance,
path=path,
impact_type=f"{change_type} propagation"
))
return ImpactAnalysisResult(
source_id=entity_id,
impacted_entities=sorted(impacted, key=lambda x: x.distance),
total_impacted=len(impacted),
risk_level=max_risk
)
def _assess_risk(self, node: LineageNode, change_type: str) -> RiskLevel:
"""根据节点类型和变更类型评估风险"""
if "compliance" in node.properties.get("tags", ""):
return RiskLevel.CRITICAL
if node.type in (NodeType.MODEL, NodeType.WORKFLOW):
return RiskLevel.HIGH
if change_type == "SCHEMA_CHANGE":
return RiskLevel.MEDIUM
return RiskLevel.LOW
#8. 血缘追踪的实际应用场景
#8.1 数据质量监控
Code
数据质量 + 血缘联动
================================================
场景: 自动检测数据质量问题的传播
质量检查失败:
┌──────────────────────────────────────┐
│ Dataset: customer_addresses │
│ 检查: 地址完整性 │
│ 结果: 失败 (23% 记录缺少邮编) │
│ 时间: 2024-03-20 08:00 │
└──────────────────┬───────────────────┘
│ 自动触发影响分析
▼
┌──────────────────────────────────────┐
│ 影响报告 │
│ │
│ 直接影响: │
│ ├─ customer_geo_analysis (Transform) │
│ ├─ delivery_routing (Model) │
│ └─ regional_revenue (Report) │
│ │
│ 间接影响: │
│ ├─ territory_planning (App) │
│ └─ logistics_optimization (Model) │
│ │
│ 建议操作: │
│ 1. 暂停 delivery_routing 模型重训 │
│ 2. 在 regional_revenue 报表标注警告 │
│ 3. 通知物流团队数据质量问题 │
└──────────────────────────────────────┘
#8.2 GDPR 合规
Code
GDPR 数据主体请求 + 血缘
================================================
场景: 用户请求删除个人数据 (Right to be Forgotten)
请求: "请删除用户 user-12345 的所有数据"
血缘追踪:
user-12345 的数据出现在:
1. CRM.customers (源)
└─ user_profile (清洗后)
├─ customer_360 (Ontology 对象)
│ ├─ customer_name: "张三"
│ ├─ customer_email: "zhang@example.com"
│ └─ customer_phone: "+86-xxx"
│
├─ marketing_targets (聚合数据集)
│ └─ 包含 user-12345 的行为数据
│
└─ ml_training_data (模型训练集)
└─ 包含 user-12345 的特征
需要删除/匿名化的位置: 4 个数据集 + 1 个 Ontology 对象
需要重训的模型: 1 个
合规记录: 记录删除操作的完整审计日志
#9. 工作流血缘
#9.1 跨系统血缘追踪
Code
工作流级血缘 (Workflow Lineage)
================================================
一个完整的数据处理工作流可能跨越多个系统:
┌─────────────────────────────────────────────────────┐
│ Workflow: "Daily Revenue Pipeline" │
│ ID: WF-2024-0320-001 │
│ │
│ Step 1: Data Ingestion (Kafka → Raw Storage) │
│ ┌─────────────────────────────────────┐ │
│ │ Source: ERP CDC Stream │ │
│ │ Target: raw.sales_orders │ │
│ │ Records: 15,234 │ │
│ │ Duration: 2m 15s │ │
│ └──────────────────────┬──────────────┘ │
│ │ │
│ Step 2: Data Cleaning (Spark Transform) │
│ ┌─────────────────────────────────────┐ │
│ │ Input: raw.sales_orders │ │
│ │ Output: clean.sales_orders │ │
│ │ Removed: 123 duplicates, 45 nulls │ │
│ │ Duration: 5m 42s │ │
│ └──────────────────────┬──────────────┘ │
│ │ │
│ Step 3: Currency Conversion (Python Transform) │
│ ┌─────────────────────────────────────┐ │
│ │ Inputs: clean.sales_orders, │ │
│ │ ref.exchange_rates │ │
│ │ Output: enriched.orders_usd │ │
│ │ Duration: 1m 03s │ │
│ └──────────────────────┬──────────────┘ │
│ │ │
│ Step 4: Ontology Sync (Ontology Updater) │
│ ┌─────────────────────────────────────┐ │
│ │ Input: enriched.orders_usd │ │
│ │ Target: Order objects (Ontology) │ │
│ │ Created: 234, Updated: 1,890 │ │
│ │ Duration: 3m 21s │ │
│ └─────────────────────────────────────┘ │
│ │
│ Total Duration: 12m 21s │
│ Status: SUCCESS │
└─────────────────────────────────────────────────────┘
#10. 构建数据信任
#10.1 信任链
数据血缘的最终目标不仅是技术能力，而是建立数据信任：
Code
数据信任金字塔
================================================
/\
/ \
/ 决策 \ "我可以基于这个数据做决策"
/ 信任 \
/──────────\
/ 数据理解 \ "我理解这个数据的含义和局限"
/──────────────\
/ 来源透明 \ "我知道数据从哪来、怎么处理的"
/────────────────────\
/ 质量保证 \ "数据经过了质量检查"
/──────────────────────────\
/ 可追溯性 \ "每一步都有记录"
/──────────────────────────────\
没有血缘 = 金字塔的地基缺失 = 无法建立信任
#Key Takeaways
-
数据血缘是连接"技术数据"和"业务信任"的桥梁 -- 它让非技术用户也能理解数据的来龙去脉，从而建立对数据驱动决策的信心。没有血缘追踪的数据平台就像没有食品溯源的超市。
-
字段级血缘是真正有价值的差异化能力 -- 实体级血缘只能告诉你"哪些表有关系"，而字段级血缘能告诉你"这个数字是怎么算出来的"。Foundry 通过代码静态分析 + 运行时捕获实现了精确的字段级血缘。
-
coomia-dip 的 LineageService 通过 11 个 RPC 接口覆盖了血缘管理的完整生命周期 -- 包括实体级和字段级血缘记录、上下游追溯、影响分析和工作流血缘，在开源方案中达到了接近 Foundry 的能力水平。
#下一篇预告
第 15 篇：Palantir 的搜索与发现 -- 在百万对象中找到你要的那个
当你的 Ontology 中有上百万个对象时，如何快速找到你需要的那个？我们将深入探讨 Foundry 的搜索引擎如何理解 Ontology 语义，提供智能、权限感知的搜索体验。
Tags: palantir data-lineage foundry field-lineage impact-analysis data-governance coomia-dip openlineage data-quality compliance