知识图谱在企业数据治理中的应用

在数字化转型的浪潮中，数据已成为企业最重要的战略资产之一。然而，随着企业业务的快速发展，数据来源日益多样、数据量呈指数级增长、数据关系日趋复杂，传统的数据治理方式已难以满足现代企业的需求。**知识图谱（Knowledge Graph）**作为一种高效的知识组织和推理工具，正在成为企业数据治理领域的重要基础设施。本文将深入探讨知识图谱如何赋能企业数据治理，从核心挑战、解决方案到具体实现进行完整剖析。

一、企业数据治理的核心挑战

1.1 数据治理的定义与范围

**数据治理（Data Governance）**是指企业对其数据资产进行全面管理的一套体系，涵盖数据的采集、存储、加工、流通、应用和销毁的全生命周期管理。成熟的数据治理体系需要解决以下核心问题：

  • 数据质量：准确性、完整性、一致性、时效性
  • 数据安全：访问控制、隐私保护、合规性
  • 数据架构：数据模型设计、系统集成、技术标准
  • 数据血缘：数据流转追踪、影响分析
  • 主数据管理：核心业务实体的统一视图
  • 元数据管理：数据的描述信息管理

1.2 传统数据治理的痛点

传统数据治理主要依赖以下技术手段：

技术方案 解决的问题 主要缺陷
关系型数据库 结构化存储 难以表达复杂关系，扩展性差
数据字典/文档 元数据管理 维护成本高，难以保持同步
ETL 作业 数据流转 血缘信息分散，难以全局追踪
数据目录 数据发现 仅提供静态描述，缺乏关联分析
规则引擎 数据质量 规则难以复用，跨系统协调困难

这些方案在实际应用中存在以下共性问题：

  1. 数据孤岛严重：各系统独立维护数据定义，缺乏全局视角
  2. 血缘断裂：数据在系统间流转时，血缘信息丢失或不一致
  3. 变更影响未知：修改数据模型时，无法快速评估影响范围
  4. 检索效率低下：基于关键词的检索难以理解语义
  5. 响应速度慢：跨系统的数据问题定位耗时且容易出错

1.3 知识图谱带来的变革

知识图谱以**三元组（Triple）**作为基本数据单元，天然适合表达企业数据环境中的复杂关联：

1
2
3
4
5
6
(数据表, 属于, 业务系统A)
(数据表, 字段, customer_id)
(数据表, 上游依赖, 数据表B)
(数据表B, 下游影响, 数据表C)
(字段, 数据类型, VARCHAR(50))
(字段, 数据Owner, 数据团队)

这种表示方式带来了显著优势：

  • 全局可见性：将企业数据资产统一建模为图谱
  • 关系驱动：支持任意维度的关系查询和分析
  • 血缘追溯：完整追踪数据从源头到消费的全链路
  • 语义理解：支持基于语义的智能检索和问答
  • 影响分析：快速评估变更对下游的影响范围
  • 智能推理：发现隐藏的数据关系和潜在问题

二、知识图谱作为数据基础设施

2.1 数据治理知识图谱的构建

企业数据治理知识图谱通常包含以下核心实体类型和关系：

实体类型：

  • 数据表/文件：物理或逻辑的数据存储单元
  • 字段/列：数据表的属性定义
  • 业务系统：数据的生产者和消费者
  • 数据模型：表结构、约束、索引等定义
  • 指标/度量：业务口径定义
  • 报表/应用：数据消费端
  • 人员/团队：数据Owner和消费者
  • 流程/作业：数据处理逻辑

关系类型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# 定义数据治理图谱的核心实体和关系
ENTITIES = {
    "DataTable": "数据表",
    "DataColumn": "数据字段", 
    "BusinessSystem": "业务系统",
    "DataModel": "数据模型",
    "Metric": "数据指标",
    "Report": "报表",
    "Application": "应用",
    "Person": "人员",
    "Team": "团队",
    "DataJob": "数据作业",
    "DataQualityRule": "数据质量规则"
}

RELATIONS = {
    # 归属关系
    ("DataTable", "belongs_to", "BusinessSystem"),
    ("DataColumn", "belongs_to", "DataTable"),
    ("DataJob", "belongs_to", "BusinessSystem"),
    
    # 血缘关系
    ("DataTable", "upstream_of", "DataTable"),  # 表级上游
    ("DataColumn", "feeds_into", "DataColumn"),  # 字段级上游
    ("DataJob", "extracts_from", "DataTable"),
    ("DataJob", "loads_into", "DataTable"),
    
    # 影响关系
    ("DataTable", "impacts", "DataTable"),  # 下游影响
    ("DataTable", "used_by", "Application"),
    
    # 语义关系
    ("DataColumn", "same_as", "DataColumn"),  # 跨系统同义字段
    ("Metric", "derived_from", "DataColumn"),
    
    # 管理关系
    ("DataTable", "owned_by", "Person"),
    ("DataTable", "stewarded_by", "Team"),
    ("DataColumn", "has_quality_rule", "DataQualityRule")
}

2.2 图谱构建的核心流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import hashlib

class EntityType(Enum):
    DATA_TABLE = "DataTable"
    DATA_COLUMN = "DataColumn"
    BUSINESS_SYSTEM = "BusinessSystem"
    DATA_MODEL = "DataModel"
    METRIC = "Metric"
    REPORT = "Report"
    APPLICATION = "Application"
    PERSON = "Person"
    TEAM = "Team"
    DATA_JOB = "DataJob"
    DATA_QUALITY_RULE = "DataQualityRule"


class RelationType(Enum):
    BELONGS_TO = "belongs_to"
    UPSTREAM_OF = "upstream_of"
    DOWNSTREAM_OF = "downstream_of"
    FEEDS_INTO = "feeds_into"
    IMPACTS = "impacts"
    USED_BY = "used_by"
    OWNED_BY = "owned_by"
    SAME_AS = "same_as"
    DERIVED_FROM = "derived_from"
    HAS_QUALITY_RULE = "has_quality_rule"


@dataclass
class KGEntity:
    """图谱实体"""
    id: str
    type: EntityType
    name: str
    properties: Dict = field(default_factory=dict)
    
    @staticmethod
    def generate_id(entity_type: EntityType, name: str) -> str:
        """生成实体唯一ID"""
        raw = f"{entity_type.value}:{name}"
        return hashlib.md5(raw.encode()).hexdigest()[:12]


@dataclass
class KGRelation:
    """图谱关系"""
    source_id: str
    target_id: str
    relation_type: RelationType
    properties: Dict = field(default_factory=dict)


class DataGovernanceKGBuilder:
    """数据治理知识图谱构建器"""
    
    def __init__(self, kg_client):
        self.kg_client = kg_client
        self.entities: Dict[str, KGEntity] = {}
        self.relations: List[KGRelation] = []
        
        # 用于去重
        self._entity_set = set()
        self._relation_set = set()
    
    def add_entity(self, entity: KGEntity) -> None:
        """添加实体到图谱"""
        entity_key = f"{entity.type.value}:{entity.name}"
        if entity_key not in self._entity_set:
            self.entities[entity.id] = entity
            self._entity_set.add(entity_key)
            
            # 同步到图数据库
            self._persist_entity(entity)
    
    def add_relation(self, relation: KGRelation) -> None:
        """添加关系到图谱"""
        relation_key = f"{relation.source_id}-{relation.relation_type.value}-{relation.target_id}"
        if relation_key not in self._relation_set:
            self.relations.append(relation)
            self._relation_set.add(relation_key)
            
            # 同步到图数据库
            self._persist_relation(relation)
    
    def _persist_entity(self, entity: KGEntity) -> None:
        """持久化实体到图数据库"""
        cypher = f"""
        MERGE (e:{entity.type.value} {{id: $id, name: $name}})
        SET e += $properties
        """
        self.kg_client.execute(cypher, {
            "id": entity.id,
            "name": entity.name,
            "properties": entity.properties
        })
    
    def _persist_relation(self, relation: KGRelation) -> None:
        """持久化关系到图数据库"""
        # 获取源和目标实体类型
        source_type = self.entities[relation.source_id].type.value
        target_type = self.entities[relation.target_id].type.value
        
        cypher = f"""
        MATCH (source:{source_type} {{id: $source_id}})
        MATCH (target:{target_type} {{id: $target_id}})
        MERGE (source)-[r:{relation.relation_type.value}]->(target)
        SET r += $properties
        """
        self.kg_client.execute(cypher, {
            "source_id": relation.source_id,
            "target_id": relation.target_id,
            "properties": relation.properties
        })

三、数据血缘：核心能力详解

3.1 数据血缘的概念与价值

**数据血缘（Data Lineage）**是指数据从源头到终点的完整流转路径记录，包含了数据在各个环节的转换、聚合、拆分等操作信息。数据血缘是数据治理的核心能力，其价值体现在：

  • 影响分析：变更数据模型时，快速定位受影响的所有下游系统
  • 问题溯源：数据质量问题发生时，追踪问题根源
  • 合规审计：满足数据监管的溯源要求
  • 指标溯源：追溯指标计算的完整数据来源
  • 数据质量追踪：评估数据处理环节对质量的影响

3.2 数据血缘的表示模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@dataclass
class DataLineageNode:
    """血缘节点"""
    node_id: str
    node_type: str  # table, column, job, file, api等
    name: str
    system: str
    properties: Dict = field(default_factory=dict)


@dataclass
class DataLineageEdge:
    """血缘边"""
    source: str  # 源节点ID
    target: str  # 目标节点ID
    transform_type: str  # direct, aggregation, join, filter, etc.
    transform_detail: str  # 详细转换逻辑
    properties: Dict = field(default_factory=dict)


class DataLineageModel:
    """数据血缘模型"""
    
    # 转换类型枚举
    TRANSFORM_TYPES = {
        "DIRECT": "直接传递",
        "FILTER": "数据过滤",
        "AGGREGATE": "聚合计算",
        "JOIN": "表关联",
        "UNION": "数据合并",
        "CALCULATE": "字段计算",
        "TYPE_CONVERT": "类型转换",
        "RENAME": "字段重命名",
        "ENRICH": "数据 enrichment"
    }
    
    @staticmethod
    def create_table_lineage(
        source_table: str,
        target_table: str,
        job_name: str,
        column_mappings: List[Tuple[str, str]] = None
    ) -> Tuple[DataLineageEdge, List[DataLineageEdge]]:
        """
        创建表级和字段级血缘
        
        Args:
            source_table: 源表
            target_table: 目标表
            job_name: ETL作业名
            column_mappings: 字段映射关系 [(源字段, 目标字段), ...]
        
        Returns:
            表级血缘边, 字段级血缘边列表
        """
        # 表级血缘
        table_edge = DataLineageEdge(
            source=source_table,
            target=target_table,
            transform_type="DIRECT",
            transform_detail=f"通过 {job_name} 处理",
            properties={"job": job_name}
        )
        
        # 字段级血缘
        column_edges = []
        if column_mappings:
            for src_col, tgt_col in column_mappings:
                column_edges.append(DataLineageEdge(
                    source=f"{source_table}.{src_col}",
                    target=f"{target_table}.{tgt_col}",
                    transform_type="DIRECT",
                    transform_detail=f"字段映射: {src_col} -> {tgt_col}",
                    properties={"source_column": src_col, "target_column": tgt_col}
                ))
        
        return table_edge, column_edges

3.3 自动血缘发现

在实际生产环境中，手工维护血缘几乎不可能。需要通过自动血缘发现技术从 SQL 脚本、ETL 配置、数据处理代码中自动提取血缘关系：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import re
from typing import List, Set, Tuple
from sqlparse import parse as parse_sql
from sqlparse.sql import Statement, Identifier, Function, Parenthesis


class SQLLineageExtractor:
    """SQL血缘自动提取器"""
    
    def __init__(self):
        self.source_tables: Set[str] = set()
        self.target_tables: Set[str] = set()
        self.column_lineage: List[Tuple[str, str]] = []  # (source, target)
    
    def extract(self, sql: str) -> Dict:
        """
        从SQL语句提取血缘关系
        
        支持的SQL类型：
        - INSERT ... SELECT
        - CREATE TABLE ... AS SELECT
        - MERGE INTO
        - UPDATE (简化处理)
        """
        statements = parse_sql(sql)
        
        for stmt in statements:
            stmt_type = stmt.get_type()
            
            if stmt_type == "INSERT":
                self._extract_insert_lineage(stmt)
            elif stmt_type == "CREATE":
                self._extract_create_lineage(stmt)
            elif stmt_type == "SELECT":
                self._extract_select_lineage(stmt)
        
        return {
            "source_tables": list(self.source_tables),
            "target_tables": list(self.target_tables),
            "column_lineage": self.column_lineage
        }
    
    def _extract_insert_lineage(self, stmt) -> None:
        """提取 INSERT 语句血缘"""
        # 解析 INSERT INTO target_table ...
        insert_match = re.search(r'INSERT\s+INTO\s+(\w+)', stmt.value, re.IGNORECASE)
        if insert_match:
            target_table = insert_match.group(1)
            self.target_tables.add(target_table)
        
        # 提取 FROM 子句中的源表
        from_match = re.search(r'FROM\s+([\w,\s\.]+)', stmt.value, re.IGNORECASE)
        if from_match:
            source_tables = [t.strip() for t in from_match.group(1).split(',')]
            self.source_tables.update(source_tables)
        
        # 提取字段映射
        self._extract_column_mapping(stmt)
    
    def _extract_create_lineage(self, stmt) -> None:
        """提取 CREATE TABLE ... AS SELECT 血缘"""
        create_match = re.search(r'CREATE\s+TABLE\s+(\w+)', stmt.value, re.IGNORECASE)
        if create_match:
            target_table = create_match.group(1)
            self.target_tables.add(target_table)
        
        # 提取 SELECT 中的源表
        select_match = re.search(r'AS\s+(SELECT\s+.+)', stmt.value, re.IGNORECASE | re.DOTALL)
        if select_match:
            select_part = select_match.group(1)
            # 简化：实际应完整解析 SELECT 语句
            from_match = re.search(r'FROM\s+([\w,\s\.]+)', select_part, re.IGNORECASE)
            if from_match:
                source_tables = [t.strip() for t in from_match.group(1).split(',')]
                self.source_tables.update(source_tables)
    
    def _extract_column_mapping(self, stmt) -> None:
        """提取字段级血缘"""
        # 简化实现：实际需要完整解析 SELECT 列表和 INSERT 列名
        # 这里演示基本逻辑
        
        # 查找 SELECT 列表
        select_pattern = r'SELECT\s+(.+?)\s+FROM'
        select_match = re.search(select_pattern, stmt.value, re.IGNORECASE | re.DOTALL)
        
        if select_match:
            select_list = select_match.group(1)
            # 解析字段列表
            columns = [c.strip() for c in select_list.split(',')]
            
            # 查找 INSERT 目标列
            insert_pattern = r'\(([^)]+)\)\s+SELECT'
            insert_match = re.search(insert_pattern, stmt.value, re.IGNORECASE)
            
            if insert_match:
                target_cols = [c.strip() for c in insert_match.group(1).split(',')]
                
                # 建立映射关系（简化：按顺序对应）
                for i, col in enumerate(columns):
                    if i < len(target_cols):
                        # 清理字段名
                        clean_col = col.split('.')[-1].strip()
                        clean_target = target_cols[i].strip()
                        self.column_lineage.append((clean_col, clean_target))
    
    def extract_from_file(self, sql_content: str) -> Dict:
        """从文件内容批量提取血缘"""
        all_lineage = {
            "source_tables": set(),
            "target_tables": set(),
            "column_lineage": []
        }
        
        # 按分号分割多条SQL
        statements = sql_content.split(';')
        
        for sql in statements:
            if sql.strip():
                self.reset()
                lineage = self.extract(sql)
                all_lineage["source_tables"].update(lineage["source_tables"])
                all_lineage["target_tables"].update(lineage["target_tables"])
                all_lineage["column_lineage"].extend(lineage["column_lineage"])
        
        return {
            "source_tables": list(all_lineage["source_tables"]),
            "target_tables": list(all_lineage["target_tables"]),
            "column_lineage": all_lineage["column_lineage"]
        }
    
    def reset(self) -> None:
        """重置提取器状态"""
        self.source_tables.clear()
        self.target_tables.clear()
        self.column_lineage.clear()

3.4 血缘查询与分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class LineageAnalyzer:
    """血缘分析器"""
    
    def __init__(self, kg_client):
        self.kg_client = kg_client
    
    def find_upstream(
        self, 
        table_name: str, 
        depth: int = None,
        include_column_level: bool = True
    ) -> List[Dict]:
        """
        查找上游血缘
        
        Args:
            table_name: 目标表名
            depth: 追溯深度，None表示完整追溯
            include_column_level: 是否包含字段级血缘
        
        Returns:
            上游血缘链路列表
        """
        if depth == 1:
            # 单层追溯
            cypher = f"""
            MATCH path = (source)-[r:UPSTREAM_OF|FEEDS_INTO*1..1]->(target:DataTable {{name: $table_name}})
            RETURN path
            """
        else:
            # 完整追溯（带深度限制防止环回）
            cypher = f"""
            MATCH path = (source)-[r:UPSTREAM_OF|FEEDS_INTO*1..5]->(target:DataTable {{name: $table_name}})
            RETURN path
            ORDER BY length(path)
            """
        
        results = self.kg_client.execute(cypher, {"table_name": table_name})
        
        lineage_paths = []
        for record in results:
            path = record['path']
            lineage_paths.append(self._format_path(path))
        
        return lineage_paths
    
    def find_downstream(
        self, 
        table_name: str, 
        depth: int = None
    ) -> List[Dict]:
        """
        查找下游影响
        
        Args:
            table_name: 源表名
            depth: 追溯深度
        """
        if depth:
            cypher = f"""
            MATCH path = (source:DataTable {{name: $table_name}})-[r:DOWNSTREAM_OF|IMPACTS*1..{depth}]->(target)
            RETURN path
            """
        else:
            cypher = """
            MATCH path = (source:DataTable {name: $table_name})-[r:DOWNSTREAM_OF|IMPACTS*1..5]->(target)
            RETURN path
            ORDER BY length(path)
            """
        
        results = self.kg_client.execute(cypher, {"table_name": table_name})
        
        impact_paths = []
        for record in results:
            path = record['path']
            impact_paths.append(self._format_path(path))
        
        return impact_paths
    
    def find_impacted_reports(self, table_name: str) -> List[str]:
        """查找受影响的报表"""
        cypher = """
        MATCH (t:DataTable {name: $table_name})-[:IMPACTS|DOWNSTREAM_OF*]->(r:Report)
        RETURN DISTINCT r.name as report_name
        """
        results = self.kg_client.execute(cypher, {"table_name": table_name})
        return [r['report_name'] for r in results]
    
    def find_root_cause(self, table_name: str) -> List[str]:
        """追溯问题根源（找到没有上游的根节点）"""
        cypher = """
        MATCH path = (root)-[r:UPSTREAM_OF*]->(target:DataTable {name: $table_name})
        WHERE NOT (root)-[:UPSTREAM_OF]->()
        RETURN root.name as root_table, length(path) as distance
        ORDER BY distance
        """
        results = self.kg_client.execute(cypher, {"table_name": table_name})
        return [r['root_table'] for r in results]
    
    def _format_path(self, path) -> Dict:
        """格式化路径为可读形式"""
        nodes = []
        relationships = []
        
        for i, element in enumerate(path):
            if hasattr(element, 'labels'):
                # Node
                labels = list(element.labels)
                nodes.append({
                    "id": element.get('id'),
                    "type": labels[0] if labels else "Unknown",
                    "name": element.get('name')
                })
            else:
                # Relationship
                relationships.append({
                    "type": type(element).__name__,
                    "properties": dict(element)
                })
        
        return {
            "nodes": nodes,
            "relationships": relationships
        }

四、主数据管理

4.1 主数据的概念与挑战

**主数据（Master Data）**是指企业核心业务实体（如客户、产品、供应商、员工）的统一定义，是跨系统共享使用的关键数据。与交易数据不同，主数据具有以下特征：

  • 共享性：被多个业务系统共享使用
  • 事务性：参与业务过程但不记录业务事件
  • 持久性：长期存在，不会因交易完成而消失
  • 参照性：作为业务交易的参照和依据

主数据管理的挑战包括：

  1. 数据分散：同一实体在不同系统中有不同表示
  2. 数据冲突：跨系统的数据定义不一致
  3. 数据冗余：重复存储导致的一致性问题
  4. 变更影响：主数据变更影响多个下游系统

4.2 基于知识图谱的主数据管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class MasterDataManagement:
    """基于知识图谱的主数据管理"""
    
    def __init__(self, kg_client):
        self.kg_client = kg_client
        
        # 主数据实体类型
        self.md_entity_types = [
            "Customer", "Product", "Supplier", 
            "Employee", "Organization", "Location"
        ]
    
    def register_master_entity(
        self,
        entity_type: str,
        canonical_name: str,
        alternative_names: List[str] = None,
        source_systems: List[str] = None,
        attributes: Dict = None
    ) -> str:
        """
        注册主数据实体
        
        Args:
            entity_type: 实体类型 (Customer, Product等)
            canonical_name: 权威名称
            alternative_names: 别名列表
            source_systems: 来源系统列表
            attributes: 属性字典
        """
        entity_id = KGEntity.generate_id(
            EntityType[entity_type.upper()], 
            canonical_name
        )
        
        # 创建权威实体
        entity = KGEntity(
            id=entity_id,
            type=EntityType[entity_type.upper()],
            name=canonical_name,
            properties={
                "is_canonical": True,
                "source_systems": source_systems or [],
                **(attributes or {})
            }
        )
        
        # 添加到图谱
        self._add_master_entity(entity)
        
        # 添加同义关系
        if alternative_names:
            for alt_name in alternative_names:
                self._add_alias(entity_id, entity_type, alt_name)
        
        # 关联来源系统
        if source_systems:
            for system in source_systems:
                self._link_to_source_system(entity_id, entity_type, system)
        
        return entity_id
    
    def _add_master_entity(self, entity: KGEntity) -> None:
        """添加主数据实体"""
        cypher = f"""
        MERGE (e:{entity.type.value} {{
            id: $id, 
            name: $name, 
            is_canonical: $is_canonical
        }})
        SET e += $properties
        """
        self.kg_client.execute(cypher, {
            "id": entity.id,
            "name": entity.name,
            "is_canonical": entity.properties.get("is_canonical", True),
            "properties": entity.properties
        })
    
    def _add_alias(self, entity_id: str, entity_type: str, alias: str) -> None:
        """添加别名关系"""
        alias_id = KGEntity.generate_id(EntityType[entity_type.upper()], alias)
        
        # 创建别名节点
        cypher = f"""
        MERGE (alias:Alias {{id: $alias_id, name: $alias}})
        MERGE (canonical:{entity_type} {{id: $entity_id}})
        MERGE (alias)-[:ALTERNATIVE_NAME_OF]->(canonical)
        """
        self.kg_client.execute(cypher, {
            "alias_id": alias_id,
            "alias": alias,
            "entity_id": entity_id
        })
    
    def _link_to_source_system(
        self, 
        entity_id: str, 
        entity_type: str, 
        system_name: str
    ) -> None:
        """关联来源系统"""
        cypher = f"""
        MATCH (md:{entity_type} {{id: $entity_id}})
        MERGE (system:BusinessSystem {{name: $system_name}})
        MERGE (md)-[:SOURCED_FROM]->(system)
        """
        self.kg_client.execute(cypher, {
            "entity_id": entity_id,
            "system_name": system_name
        })
    
    def find_master_entity(
        self, 
        entity_type: str, 
        search_term: str
    ) -> List[Dict]:
        """
        查找主数据实体（支持模糊匹配）
        
        通过别名也能找到主实体
        """
        cypher = f"""
        // 直接匹配
        MATCH (md:{entity_type} {{name: $search_term}})
        WHERE md.is_canonical = true OR md.is_canonical IS NULL
        RETURN md, 'direct' as match_type
        
        UNION
        
        // 通过别名匹配
        MATCH (alias:Alias {{name: $search_term}})-[:ALTERNATIVE_NAME_OF]->(md:{entity_type})
        RETURN md, 'alias' as match_type
        """
        results = self.kg_client.execute(cypher, {"search_term": search_term})
        return [dict(r['md']) for r in results]
    
    def get_entity_consistency_report(self, entity_type: str) -> Dict:
        """
        生成数据一致性报告
        识别同一主实体在不同系统中的数据冲突
        """
        cypher = f"""
        MATCH (md:{entity_type})
        WHERE md.is_canonical = true
        MATCH (md)-[:SOURCED_FROM]->(system:BusinessSystem)
        OPTIONAL MATCH (md)-[:HAS_ATTRIBUTE]->(attr)
        RETURN md.id as entity_id,
               md.name as entity_name,
               collect(DISTINCT system.name) as source_systems,
               md
        """
        results = self.kg_client.execute(cypher)
        
        report = {
            "total_entities": len(results),
            "multi_source_entities": 0,
            "entities": []
        }
        
        for r in results:
            if len(r['source_systems']) > 1:
                report["multi_source_entities"] += 1
            
            report["entities"].append({
                "id": r['entity_id'],
                "name": r['entity_name'],
                "sources": r['source_systems'],
                "is_multi_source": len(r['source_systems']) > 1
            })
        
        return report

4.3 主数据匹配与合并

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
class MDMEntityResolver:
    """主数据实体解析与匹配"""
    
    def __init__(self, similarity_threshold: float = 0.85):
        self.similarity_threshold = similarity_threshold
        self.embedding_model = None  # 可使用文本嵌入模型
    
    def find_duplicates(
        self, 
        entity_type: str, 
        entity_records: List[Dict]
    ) -> List[List[str]]:
        """
        查找潜在重复的主数据实体
        
        Args:
            entity_type: 实体类型
            entity_records: 待检测的实体记录列表
        
        Returns:
            重复实体ID分组
        """
        duplicates = []
        processed = set()
        
        for i, record1 in enumerate(entity_records):
            if record1['id'] in processed:
                continue
            
            same_group = [record1['id']]
            
            for j, record2 in enumerate(entity_records[i+1:], i+1):
                if record2['id'] in processed:
                    continue
                
                similarity = self._calculate_similarity(record1, record2)
                
                if similarity >= self.similarity_threshold:
                    same_group.append(record2['id'])
                    processed.add(record2['id'])
            
            if len(same_group) > 1:
                duplicates.append(same_group)
                processed.add(record1['id'])
        
        return duplicates
    
    def _calculate_similarity(
        self, 
        record1: Dict, 
        record2: Dict
    ) -> float:
        """
        计算两条记录的相似度
        
        支持多种策略：
        - 字段级相似度
        - 文本嵌入相似度
        - 规则匹配
        """
        if record1.get('name') == record2.get('name'):
            return 1.0
        
        # 名称相似度（简单实现）
        name_sim = self._string_similarity(
            record1.get('name', ''), 
            record2.get('name', '')
        )
        
        # 属性相似度
        attr_sim = self._attribute_similarity(
            record1.get('attributes', {}),
            record2.get('attributes', {})
        )
        
        # 加权平均
        return 0.7 * name_sim + 0.3 * attr_sim
    
    def _string_similarity(self, s1: str, s2: str) -> float:
        """字符串相似度（使用Jaccard）"""
        set1 = set(s1.lower())
        set2 = set(s2.lower())
        intersection = len(set1 & set2)
        union = len(set1 | set2)
        return intersection / union if union > 0 else 0
    
    def _attribute_similarity(self, attrs1: Dict, attrs2: Dict) -> float:
        """属性相似度"""
        if not attrs1 and not attrs2:
            return 1.0
        if not attrs1 or not attrs2:
            return 0.0
        
        common_keys = set(attrs1.keys()) & set(attrs2.keys())
        if not common_keys:
            return 0.0
        
        matches = sum(
            1 for k in common_keys 
            if attrs1[k] == attrs2[k]
        )
        
        return matches / len(common_keys)

五、数据质量管理与智能问答

5.1 基于知识图谱的数据质量管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class DataQualityManager:
    """基于知识图谱的数据质量管理"""
    
    def __init__(self, kg_client):
        self.kg_client = kg_client
    
    def register_quality_rule(
        self,
        table_name: str,
        column_name: str,
        rule_type: str,  # not_null, unique, range, pattern, etc.
        rule_definition: Dict,
        owner: str = None
    ) -> str:
        """注册数据质量规则"""
        rule_id = f"rule_{table_name}_{column_name}_{rule_type}"
        
        cypher = f"""
        MERGE (t:DataTable {{name: $table_name}})
        MERGE (c:DataColumn {{name: $column_name, table: $table_name}})
        MERGE (c)-[:HAS_QUALITY_RULE]->(r:DataQualityRule {{
            id: $rule_id,
            type: $rule_type,
            definition: $rule_definition
        }})
        SET r.owner = $owner
        """
        self.kg_client.execute(cypher, {
            "table_name": table_name,
            "column_name": column_name,
            "rule_id": rule_id,
            "rule_type": rule_type,
            "rule_definition": rule_definition,
            "owner": owner
        })
        
        return rule_id
    
    def check_quality_issues(self, table_name: str) -> Dict:
        """检查数据质量问题"""
        cypher = f"""
        MATCH (t:DataTable {{name: $table_name}})
        MATCH (c:DataColumn)-[:BELONGS_TO]->(t)
        MATCH (c)-[:HAS_QUALITY_RULE]->(r:DataQualityRule)
        RETURN c.name as column_name, r.type as rule_type, r.definition as rule_def
        """
        results = self.kg_client.execute(cypher, {"table_name": table_name})
        
        issues = []
        for r in results:
            # 实际执行质量检查
            issue = self._execute_quality_check(
                table_name, 
                r['column_name'], 
                r['rule_type'], 
                r['rule_def']
            )
            if issue:
                issues.append(issue)
        
        return {
            "table": table_name,
            "total_rules": len(results),
            "issues": issues
        }
    
    def _execute_quality_check(
        self, 
        table: str, 
        column: str, 
        rule_type: str, 
        rule_def: Dict
    ) -> Optional[Dict]:
        """执行质量检查"""
        # 实际实现需要查询数据源
        # 这里返回模拟结果
        return None
    
    def get_quality_dashboard(self) -> Dict:
        """生成数据质量仪表盘"""
        cypher = """
        MATCH (t:DataTable)
        MATCH (c:DataColumn)-[:BELONGS_TO]->(t)
        MATCH (c)-[:HAS_QUALITY_RULE]->(r:DataQualityRule)
        RETURN t.name as table_name, 
               count(DISTINCT c) as monitored_columns,
               count(DISTINCT r) as quality_rules
        """
        results = self.kg_client.execute(cypher)
        
        return {
            "tables": [dict(r) for r in results],
            "total_tables": len(results),
            "summary": {
                "total_columns": sum(r['monitored_columns'] for r in results),
                "total_rules": sum(r['quality_rules'] for r in results)
            }
        }

5.2 智能数据治理问答

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class DataGovernanceChatbot:
    """数据治理智能问答助手"""
    
    def __init__(self, kg_client, llm):
        self.kg_client = kg_client
        self.llm = llm
        self.question_patterns = self._init_patterns()
    
    def _init_patterns(self) -> Dict:
        """初始化问题模式"""
        return {
            "lineage_upstream": [
                r"(.*)的数据来源是",
                r"(.*)依赖于哪些表",
                r"哪些表给.*提供数据",
            ],
            "lineage_downstream": [
                r"(.*)影响了哪些",
                r"修改.*会影响哪些",
                r"(.*)的下游有哪些",
            ],
            "data_quality": [
                r"(.*)的数据质量",
                r"(.*)有什么问题",
                r"(.*)的数据质量报告",
            ],
            "master_data": [
                r"(.*)的主数据",
                r"(.*)的定义",
                r"(.*)的标准是什么",
            ],
            "impact_analysis": [
                r"如果修改(.*)会怎样",
                r"变更(.*)的影响",
            ]
        }
    
    def answer(self, question: str) -> Dict:
        """回答数据治理相关问题"""
        question_type = self._classify_question(question)
        
        if question_type == "lineage_upstream":
            return self._answer_upstream(question)
        elif question_type == "lineage_downstream":
            return self._answer_downstream(question)
        elif question_type == "data_quality":
            return self._answer_quality(question)
        elif question_type == "master_data":
            return self._answer_master_data(question)
        elif question_type == "impact_analysis":
            return self._answer_impact(question)
        else:
            return self._fallback_answer(question)
    
    def _classify_question(self, question: str) -> str:
        """问题分类"""
        for qtype, patterns in self.question_patterns.items():
            for pattern in patterns:
                if re.search(pattern, question):
                    return qtype
        return "unknown"
    
    def _answer_upstream(self, question: str) -> Dict:
        """回答上游血缘问题"""
        table_name = self._extract_table_name(question)
        if not table_name:
            return {"answer": "无法从问题中识别数据表", "confidence": 0}
        
        analyzer = LineageAnalyzer(self.kg_client)
        upstream = analyzer.find_upstream(table_name)
        
        if upstream:
            answer = f"{table_name} 的数据来源包括：\n"
            for path in upstream:
                nodes = path['nodes']
                answer += f"  - {' -> '.join([n['name'] for n in nodes])}\n"
        else:
            answer = f"未找到 {table_name} 的上游血缘信息"
        
        return {"answer": answer, "type": "lineage", "confidence": 0.9}
    
    def _answer_downstream(self, question: str) -> Dict:
        """回答下游影响问题"""
        table_name = self._extract_table_name(question)
        if not table_name:
            return {"answer": "无法从问题中识别数据表", "confidence": 0}
        
        analyzer = LineageAnalyzer(self.kg_client)
        downstream = analyzer.find_downstream(table_name)
        reports = analyzer.find_impacted_reports(table_name)
        
        answer = f"修改 {table_name} 可能影响：\n"
        
        if downstream:
            answer += "下游表：\n"
            for path in downstream:
                nodes = path['nodes']
                answer += f"  - {' -> '.join([n['name'] for n in nodes])}\n"
        
        if reports:
            answer += f"下游报表：\n"
            for report in reports:
                answer += f"  - {report}\n"
        
        if not downstream and not reports:
            answer = f"未找到 {table_name} 的下游影响信息"
        
        return {"answer": answer, "type": "lineage", "confidence": 0.9}
    
    def _answer_quality(self, question: str) -> Dict:
        """回答数据质量问题"""
        table_name = self._extract_table_name(question)
        if not table_name:
            return {"answer": "无法从问题中识别数据表", "confidence": 0}
        
        dqm = DataQualityManager(self.kg_client)
        quality_report = dqm.check_quality_issues(table_name)
        
        answer = f"{table_name} 数据质量概况：\n"
        answer += f"- 监控字段数：{quality_report['total_rules']}\n"
        answer += f"- 发现问题数：{len(quality_report['issues'])}\n"
        
        if quality_report['issues']:
            answer += "\n问题详情：\n"
            for issue in quality_report['issues']:
                answer += f"- {issue['description']}\n"
        
        return {"answer": answer, "type": "quality", "confidence": 0.85}
    
    def _answer_master_data(self, question: str) -> Dict:
        """回答主数据问题"""
        entity_name = self._extract_entity_name(question)
        if not entity_name:
            return {"answer": "无法从问题中识别主数据实体", "confidence": 0}
        
        mdm = MasterDataManagement(self.kg_client)
        entities = mdm.find_master_entity("Customer", entity_name)  # 简化：默认查Customer
        
        if entities:
            entity = entities[0]
            answer = f"主数据实体：{entity.get('name')}\n"
            answer += f"类型：{entity.get('type')}\n"
            answer += f"来源系统：{', '.join(entity.get('source_systems', []))}\n"
        else:
            answer = f"未找到主数据实体：{entity_name}"
        
        return {"answer": answer, "type": "master_data", "confidence": 0.8}
    
    def _answer_impact(self, question: str) -> Dict:
        """回答变更影响问题"""
        table_name = self._extract_table_name(question)
        if not table_name:
            return {"answer": "无法从问题中识别数据表", "confidence": 0}
        
        analyzer = LineageAnalyzer(self.kg_client)
        downstream = analyzer.find_downstream(table_name)
        reports = analyzer.find_impacted_reports(table_name)
        
        answer = f"变更影响分析报告：\n"
        answer += f"目标表：{table_name}\n\n"
        
        answer += f"直接下游：{len(downstream)} 个\n"
        answer += f"受影响报表：{len(reports)} 个\n\n"
        
        if downstream:
            answer += "影响链路：\n"
            for i, path in enumerate(downstream[:5], 1):  # 限制显示前5条
                nodes = path['nodes']
                answer += f"  {i}. {' → '.join([n['name'] for n in nodes])}\n"
        
        return {"answer": answer, "type": "impact", "confidence": 0.9}
    
    def _extract_table_name(self, question: str) -> Optional[str]:
        """从问题中提取表名"""
        # 简化实现：实际可使用NER模型
        patterns = [
            r"(?:表|table)\s*([a-zA-Z0-9_]+)",
            r"([a-zA-Z0-9_]+)\s*(?:表|table)",
            r"(?:修改|变更|影响|来源).*?([a-zA-Z0-9_]+)"
        ]
        
        for pattern in patterns:
            match = re.search(pattern, question)
            if match:
                return match.group(1)
        return None
    
    def _extract_entity_name(self, question: str) -> Optional[str]:
        """从问题中提取实体名"""
        # 简化实现
        match = re.search(r"(?:的|是什么)[:：]?\s*([^\s，。？]+)", question)
        return match.group(1) if match else None
    
    def _fallback_answer(self, question: str) -> Dict:
        """无法分类时的兜底处理"""
        return {
            "answer": f"您的问题「{question}」需要进一步分析，请尝试以下方式描述：\n"
                     f"- 「XX表的数据来源是什么」\n"
                     f"- 「XX表影响了哪些下游」\n"
                     f"- 「XX表的数据质量如何」\n"
                     f"- 「XX的主数据定义」",
            "type": "fallback",
            "confidence": 0.3
        }

六、实战：企业级数据治理平台架构

6.1 整体架构设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
┌─────────────────────────────────────────────────────────────────────┐
│                        数据治理知识图谱平台                           │
├─────────────────────────────────────────────────────────────────────┤
│  数据源层                                                              │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐       │
│  │Hive Metastore││MySQL   ││Kafka    ││API      ││File     │       │
│  │(元数据)   ││(业务库)  ││(实时流)  ││(外部)   ││(日志)   │       │
│  └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘       │
│       │           │           │           │           │            │
├───────┴───────────┴───────────┴───────────┴───────────┴────────────┤
│  采集层                                                               │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐               │
│  │元数据采集器   │ │血缘解析器    │ │质量探针      │               │
│  │(Connector)   │ │(SQL Parser) │ │(Data Profiler)│               │
│  └──────┬───────┘ └──────┬───────┘ └──────┬───────┘               │
│         │                │                │                        │
├─────────┴────────────────┴────────────────┴─────────────────────────┤
│  图谱存储层                                                           │
│  ┌─────────────────────────────────────────────────────────────┐   │
│  │                    Neo4j / JanusGraph                        │   │
│  │  (DataTable, DataColumn, BusinessSystem, DataJob, Metric)   │   │
│  └─────────────────────────────────────────────────────────────┘   │
├─────────────────────────────────────────────────────────────────────┤
│  服务层                                                               │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │血缘服务   │ │质量服务   │ │目录服务   │ │问答服务   │              │
│  │LineageAPI│ │QualityAPI│ │CatalogAPI│ │ChatBotAPI│              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
├─────────────────────────────────────────────────────────────────────┤
│  应用层                                                               │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│  │数据目录   │ │血缘地图   │ │质量监控   │ │智能问答   │              │
│  │(Portal) │ │(Lineage) │ │(Monitor) │ │(QA Chat) │              │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘              │
└─────────────────────────────────────────────────────────────────────┘

6.2 核心实现代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
from flask import Flask, request, jsonify
from functools import wraps
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 全局组件（实际应通过依赖注入）
kg_client = None  # 图数据库客户端
llm = None  # LLM模型


class DataGovernancePlatform:
    """数据治理平台主类"""
    
    def __init__(self, kg_client, llm):
        self.kg_client = kg_client
        self.llm = llm
        
        # 初始化各服务
        self.kg_builder = DataGovernanceKGBuilder(kg_client)
        self.lineage_analyzer = LineageAnalyzer(kg_client)
        self.mdm = MasterDataManagement(kg_client)
        self.dqm = DataQualityManager(kg_client)
        self.chatbot = DataGovernanceChatbot(kg_client, llm)
    
    def register_data_asset(
        self,
        asset_type: str,
        name: str,
        system: str,
        properties: Dict = None
    ) -> str:
        """注册数据资产到图谱"""
        entity_type = EntityType[asset_type.upper()]
        entity_id = KGEntity.generate_id(entity_type, name)
        
        entity = KGEntity(
            id=entity_id,
            type=entity_type,
            name=name,
            properties={
                "system": system,
                **(properties or {})
            }
        )
        
        self.kg_builder.add_entity(entity)
        
        # 关联业务系统
        system_id = KGEntity.generate_id(EntityType.BUSINESS_SYSTEM, system)
        system_entity = KGEntity(
            id=system_id,
            type=EntityType.BUSINESS_SYSTEM,
            name=system,
            properties={}
        )
        self.kg_builder.add_entity(system_entity)
        
        relation = KGRelation(
            source_id=entity_id,
            target_id=system_id,
            relation_type=RelationType.BELONGS_TO
        )
        self.kg_builder.add_relation(relation)
        
        return entity_id
    
    def register_lineage(
        self,
        source_table: str,
        target_table: str,
        job_name: str,
        column_mappings: List[Tuple[str, str]] = None
    ) -> None:
        """注册数据血缘"""
        # 创建血缘边
        table_edge, column_edges = DataLineageModel.create_table_lineage(
            source_table, target_table, job_name, column_mappings
        )
        
        # 添加到图谱
        source_id = KGEntity.generate_id(EntityType.DATA_TABLE, source_table)
        target_id = KGEntity.generate_id(EntityType.DATA_TABLE, target_table)
        
        relation = KGRelation(
            source_id=source_id,
            target_id=target_id,
            relation_type=RelationType.UPSTREAM_OF,
            properties={"job": job_name}
        )
        self.kg_builder.add_relation(relation)
        
        # 字段级血缘
        for col_edge in column_edges:
            col_source_id = f"{source_table}.{col_edge.source}"
            col_target_id = f"{target_table}.{col_edge.target}"
            
            # 简化的字段节点ID
            col_relation = KGRelation(
                source_id=col_source_id,
                target_id=col_target_id,
                relation_type=RelationType.FEEDS_INTO,
                properties={"transform": col_edge.transform_detail}
            )
            self.kg_builder.add_relation(col_relation)


# API 路由
@app.route("/api/asset/register", methods=["POST"])
def register_asset():
    """注册数据资产"""
    data = request.json
    
    platform = DataGovernancePlatform(kg_client, llm)
    asset_id = platform.register_asset(
        asset_type=data.get("type"),
        name=data.get("name"),
        system=data.get("system"),
        properties=data.get("properties", {})
    )
    
    return jsonify({"success": True, "asset_id": asset_id})


@app.route("/api/lineage/register", methods=["POST"])
def register_lineage():
    """注册数据血缘"""
    data = request.json
    
    platform = DataGovernancePlatform(kg_client, llm)
    platform.register_lineage(
        source_table=data["source_table"],
        target_table=data["target_table"],
        job_name=data.get("job_name", "unknown"),
        column_mappings=data.get("column_mappings")
    )
    
    return jsonify({"success": True})


@app.route("/api/lineage/upstream", methods=["GET"])
def get_upstream():
    """查询上游血缘"""
    table_name = request.args.get("table_name")
    depth = request.args.get("depth", type=int)
    
    analyzer = LineageAnalyzer(kg_client)
    upstream = analyzer.find_upstream(table_name, depth)
    
    return jsonify({
        "table": table_name,
        "upstream": upstream
    })


@app.route("/api/lineage/downstream", methods=["GET"])
def get_downstream():
    """查询下游影响"""
    table_name = request.args.get("table_name")
    depth = request.args.get("depth", type=int)
    
    analyzer = LineageAnalyzer(kg_client)
    downstream = analyzer.find_downstream(table_name, depth)
    reports = analyzer.find_impacted_reports(table_name)
    
    return jsonify({
        "table": table_name,
        "downstream": downstream,
        "affected_reports": reports
    })


@app.route("/api/quality/check", methods=["GET"])
def check_quality():
    """检查数据质量"""
    table_name = request.args.get("table_name")
    
    dqm = DataQualityManager(kg_client)
    report = dqm.check_quality_issues(table_name)
    
    return jsonify(report)


@app.route("/api/chat/answer", methods=["POST"])
def chat():
    """数据治理智能问答"""
    question = request.json.get("question")
    
    chatbot = DataGovernanceChatbot(kg_client, llm)
    result = chatbot.answer(question)
    
    return jsonify(result)


if __name__ == "__main__":
    # 启动服务
    # app.run(host="0.0.0.0", port=8080)
    print("数据治理平台 API 服务示例")

七、总结与展望

知识图谱为企业的数据治理带来了革命性的变化。通过将企业数据资产全面建模为图谱结构，组织能够实现：

  1. 全局数据可见性：统一视图管理分散在各系统中的数据资产
  2. 完整的血缘追踪：从数据源头到消费端的全链路追溯
  3. 智能的影响分析：快速评估变更对下游的影响范围
  4. 高效的主数据管理：跨系统的实体识别与一致性管理
  5. 智能化的数据问答：基于语义的智能数据检索与问答

未来，随着技术的进一步发展，数据治理知识图谱将更加智能化：

  • 实时血缘更新：与数据处理流程深度集成，实现血缘的实时自动更新
  • AI 增强的质量管理：利用机器学习自动发现数据质量问题模式
  • 智能问答深化：结合大语言模型，实现更自然的数据治理交互体验
  • 自动化元数据生成：利用 AI 自动提取和补全元数据信息

相关标签：知识图谱、数据治理、数据血缘、企业数据、数据管理

推荐阅读：