1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

典型企业数据资产现状：
┌────────────────────────────────────────────────────────────┐
│  · 200+ 数据库实例（MySQL/PostgreSQL/Oracle/MongoDB...）   │
│  · 5000+ 数据表（没人能说清楚每张表的用途和负责人）       │
│  · 30+ 数据仓库/数据湖中的数据集                          │
│  · 100+ API数据接口（不知道谁在用、用了什么字段）          │
│  · 数以TB计的文件数据（Excel/CSV/日志文件散落各处）        │
│                                                            │
│  核心矛盾：                                               │
│  数据越来越多，但没人知道：                                │
│  1. 到底有哪些数据？（资产盘点）                           │
│  2. 数据质量如何？（可信度）                               │
│  3. 数据值多少钱？（价值量化）                             │
└────────────────────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

元数据采集架构：
┌───────────────────────────────────────────────────────────────┐
│                      元数据管理平台                             │
│  ┌─────────────────────────────────────────────────────────┐  │
│  │                 元数据存储层（Metadata Store）             │  │
│  │  · 技术元数据 → Neo4j（图数据库，适合关系建模）          │  │
│  │  · 业务元数据 → PostgreSQL（结构化存储）                 │  │
│  │  · 操作元数据 → ClickHouse（时序分析）                   │  │
│  └───────────────────────────┬─────────────────────────────┘  │
│                              │                                │
│  ┌───────────────────────────▼─────────────────────────────┐  │
│  │              采集调度层（Apache Airflow）                 │  │
│  │  · 定时采集任务  · 增量更新  · 异常告警                  │  │
│  └───┬──────────┬──────────┬──────────┬──────────┬────────┘  │
│      │          │          │          │          │           │
│  ┌───▼───┐ ┌───▼───┐ ┌───▼───┐ ┌───▼───┐ ┌───▼───┐       │
│  │ RDBMS │ │ NoSQL │ │ Hive/ │ │ Kafka │ │ API   │       │
│  │采集器 │ │采集器 │ │Spark  │ │采集器 │ │采集器 │       │
│  │       │ │       │ │采集器 │ │       │ │       │       │
│  └───────┘ └───────┘ └───────┘ └───────┘ └───────┘       │
└───────────────────────────────────────────────────────────────┘

 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

"""关系型数据库元数据采集器"""
import psycopg2
from sqlalchemy import create_engine, inspect
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

@dataclass
class ColumnMetadata:
    table_schema: str
    table_name: str
    column_name: str
    data_type: str
    is_nullable: bool
    column_default: Optional[str]
    character_max_length: Optional[int]
    numeric_precision: Optional[int]
    column_comment: Optional[str]

@dataclass
class TableMetadata:
    table_schema: str
    table_name: str
    table_type: str  # BASE TABLE, VIEW, etc.
    row_count: int
    data_size_bytes: int
    last_modified: Optional[datetime]
    table_comment: Optional[str]
    columns: List[ColumnMetadata]
    indexes: List[dict]
    foreign_keys: List[dict]

class RDBMSMetadataCollector:
    """关系型数据库元数据采集器"""
    
    def __init__(self, connection_string: str):
        self.engine = create_engine(connection_string)
        self.inspector = inspect(self.engine)
    
    def collect_all_tables(self, schema: str = 'public') -> List[TableMetadata]:
        """采集指定schema下所有表的元数据"""
        tables = []
        
        for table_name in self.inspector.get_table_names(schema=schema):
            table_meta = self._collect_table(schema, table_name)
            tables.append(table_meta)
        
        return tables
    
    def _collect_table(self, schema: str, table_name: str) -> TableMetadata:
        """采集单张表的完整元数据"""
        # 获取列信息
        columns = []
        for col in self.inspector.get_columns(table_name, schema=schema):
            columns.append(ColumnMetadata(
                table_schema=schema,
                table_name=table_name,
                column_name=col['name'],
                data_type=str(col['type']),
                is_nullable=col.get('nullable', True),
                column_default=str(col.get('default', '')),
                character_max_length=getattr(col['type'], 'length', None),
                numeric_precision=getattr(col['type'], 'precision', None),
                column_comment=col.get('comment', '')
            ))
        
        # 获取索引信息
        indexes = self.inspector.get_indexes(table_name, schema=schema)
        
        # 获取外键信息
        foreign_keys = self.inspector.get_foreign_keys(table_name, schema=schema)
        
        # 获取表行数和大小（PostgreSQL特有）
        with self.engine.connect() as conn:
            row_count = conn.execute(
                f"SELECT reltuples::bigint FROM pg_class "
                f"WHERE relname = '{table_name}'"
            ).scalar()
            
            data_size = conn.execute(
                f"SELECT pg_total_relation_size('{schema}.{table_name}')"
            ).scalar()
            
            table_comment = conn.execute(
                f"SELECT obj_description('{schema}.{table_name}'::regclass)"
            ).scalar()
        
        return TableMetadata(
            table_schema=schema,
            table_name=table_name,
            table_type='BASE TABLE',
            row_count=row_count or 0,
            data_size_bytes=data_size or 0,
            last_modified=None,
            table_comment=table_comment,
            columns=columns,
            indexes=indexes,
            foreign_keys=foreign_keys
        )

 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

"""Hive元数据采集器 - 直接读取Hive Metastore"""
from pyhive import hive

class HiveMetadataCollector:
    def __init__(self, metastore_uri: str):
        self.conn = hive.connect(host=metastore_uri, port=10000)
    
    def collect_database(self, db_name: str) -> dict:
        """采集数据库级别的元数据"""
        cursor = self.conn.cursor()
        
        # 获取所有表
        cursor.execute(f"SHOW TABLES IN {db_name}")
        tables = [row[0] for row in cursor.fetchall()]
        
        result = {"database": db_name, "tables": []}
        
        for table in tables:
            # 获取表结构
            cursor.execute(f"DESCRIBE FORMATTED {db_name}.{table}")
            describe_rows = cursor.fetchall()
            
            # 解析DESCRIBE FORMATTED输出
            table_meta = self._parse_describe_output(table, describe_rows)
            result["tables"].append(table_meta)
        
        return result
    
    def _parse_describe_output(self, table_name: str, rows: list) -> dict:
        """解析Hive DESCRIBE FORMATTED的输出"""
        meta = {
            "name": table_name,
            "columns": [],
            "partition_keys": [],
            "storage_info": {},
            "table_params": {}
        }
        
        current_section = "columns"
        for row in rows:
            col_name = row[0].strip() if row[0] else ""
            data_type = row[1].strip() if row[1] else ""
            comment = row[2].strip() if row[2] else ""
            
            if "# col_name" in col_name:
                current_section = "columns"
                continue
            elif "# Partition Information" in col_name:
                current_section = "partition"
                continue
            elif "# Detailed Table Information" in col_name:
                current_section = "details"
                continue
            
            if current_section == "columns" and col_name and data_type:
                meta["columns"].append({
                    "name": col_name,
                    "type": data_type,
                    "comment": comment
                })
            elif current_section == "partition" and col_name:
                meta["partition_keys"].append({
                    "name": col_name,
                    "type": data_type
                })
            elif current_section == "details":
                if col_name and data_type:
                    meta["table_params"][col_name] = data_type
        
        return meta

 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

# Airflow DAG配置 - 元数据定时采集
default_args:
  owner: 'data-governance'
  start_date: '2026-06-01'
  retries: 3
  retry_delay: 5m

metadata_collection_dag:
  schedule: '0 2 * * *'  # 每天凌晨2点执行
  tasks:
    - task_id: collect_mysql_prod
      operator: PythonOperator
      python_callable: collect_rdbms_metadata
      op_kwargs:
        connection_string: "mysql+pymysql://reader:***@mysql-prod:3306"
        schemas: ["order_db", "user_db", "product_db"]
    
    - task_id: collect_hive_warehouse
      operator: PythonOperator
      python_callable: collect_hive_metadata
      op_kwargs:
        metastore_uri: "hive-metastore-01"
        databases: ["dw", "dwd", "dws", "ads"]
    
    - task_id: collect_kafka_topics
      operator: PythonOperator
      python_callable: collect_kafka_metadata
      op_kwargs:
        bootstrap_servers: "kafka-01:9092,kafka-02:9092"
    
    - task_id: merge_and_update
      operator: PythonOperator
      python_callable: merge_metadata_to_store
      dependencies: [collect_mysql_prod, collect_hive_warehouse, collect_kafka_topics]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

数据质量六维度：
┌──────────────────────────────────────────────────────────────┐
│                                                              │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐                │
│  │ 完整性   │   │ 准确性   │   │ 一致性   │                │
│  │Completnes│   │ Accuracy │   │Consistency│                │
│  │          │   │          │   │          │                │
│  │NULL率    │   │值域校验  │   │跨表一致  │                │
│  │空值率    │   │格式校验  │   │跨系统一致│                │
│  └──────────┘   └──────────┘   └──────────┘                │
│                                                              │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐                │
│  │ 及时性   │   │ 唯一性   │   │ 有效性   │                │
│  │Timelines │   │Uniqueness│   │ Validity │                │
│  │          │   │          │   │          │                │
│  │数据延迟  │   │主键唯一  │   │业务规则  │                │
│  │更新频率  │   │去重率    │   │逻辑校验  │                │
│  └──────────┘   └──────────┘   └──────────┘                │
│                                                              │
└──────────────────────────────────────────────────────────────┘

  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

"""数据质量规则引擎"""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict
from enum import Enum
import pandas as pd

class QualityDimension(Enum):
    COMPLETENESS = "完整性"
    ACCURACY = "准确性"
    CONSISTENCY = "一致性"
    TIMELINESS = "及时性"
    UNIQUENESS = "唯一性"
    VALIDITY = "有效性"

@dataclass
class QualityRule:
    rule_id: str
    rule_name: str
    dimension: QualityDimension
    table: str
    column: str
    expression: str  # SQL表达式
    threshold: float  # 阈值（0-1）
    severity: str  # CRITICAL, WARNING, INFO

@dataclass
class QualityResult:
    rule: QualityRule
    actual_value: float
    passed: bool
    sample_failures: List[dict]  # 不合格的样例数据
    execution_time: float

class QualityEngine:
    """数据质量检查引擎"""
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.rules: List[QualityRule] = []
    
    def add_rule(self, rule: QualityRule):
        self.rules.append(rule)
    
    def add_standard_rules(self, table: str, columns: List[dict]):
        """根据表结构自动生成标准质量规则"""
        for col in columns:
            # 完整性：非空字段的NULL率检查
            if not col.get('nullable', True):
                self.add_rule(QualityRule(
                    rule_id=f"COMP_{table}_{col['name']}",
                    rule_name=f"{table}.{col['name']}完整性检查",
                    dimension=QualityDimension.COMPLETENESS,
                    table=table,
                    column=col['name'],
                    expression=f"""
                        SELECT 
                            COUNT(*) - COUNT({col['name']}) AS null_count,
                            COUNT(*) AS total_count,
                            1.0 - (COUNT(*) - COUNT({col['name']}))::float / COUNT(*) AS score
                        FROM {table}
                        WHERE created_at >= NOW() - INTERVAL '1 day'
                    """,
                    threshold=0.99,
                    severity="CRITICAL"
                ))
            
            # 唯一性：主键和唯一约束字段
            if col.get('unique', False) or col.get('primary_key', False):
                self.add_rule(QualityRule(
                    rule_id=f"UNIQ_{table}_{col['name']}",
                    rule_name=f"{table}.{col['name']}唯一性检查",
                    dimension=QualityDimension.UNIQUENESS,
                    table=table,
                    column=col['name'],
                    expression=f"""
                        SELECT 
                            COUNT(DISTINCT {col['name']})::float / COUNT(*) AS score
                        FROM {table}
                    """,
                    threshold=1.0,
                    severity="CRITICAL"
                ))
            
            # 准确性：值域检查（针对有枚举范围的字段）
            if col.get('valid_values'):
                values = col['valid_values']
                self.add_rule(QualityRule(
                    rule_id=f"ACCU_{table}_{col['name']}",
                    rule_name=f"{table}.{col['name']}值域检查",
                    dimension=QualityDimension.ACCURACY,
                    table=table,
                    column=col['name'],
                    expression=f"""
                        SELECT 
                            COUNT(CASE WHEN {col['name']} IN ({values}) THEN 1 END)::float 
                            / COUNT(*) AS score
                        FROM {table}
                        WHERE created_at >= NOW() - INTERVAL '1 day'
                    """,
                    threshold=0.95,
                    severity="WARNING"
                ))
    
    def execute_all(self) -> List[QualityResult]:
        """执行所有质量规则"""
        results = []
        for rule in self.rules:
            result = self._execute_rule(rule)
            results.append(result)
        return results
    
    def _execute_rule(self, rule: QualityRule) -> QualityResult:
        """执行单条质量规则"""
        import time
        start = time.time()
        
        cursor = self.db.cursor()
        cursor.execute(rule.expression)
        row = cursor.fetchone()
        
        score = row[-1]  # 最后一列是score
        execution_time = time.time() - start
        
        # 获取不合格样例
        sample_failures = []
        if score < rule.threshold:
            sample_query = self._build_failure_sample_query(rule)
            cursor.execute(sample_query)
            sample_failures = [dict(zip(
                [desc[0] for desc in cursor.description], row
            )) for row in cursor.fetchmany(10)]
        
        return QualityResult(
            rule=rule,
            actual_value=score,
            passed=score >= rule.threshold,
            sample_failures=sample_failures,
            execution_time=execution_time
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

数据质量评分卡示例（order_db.orders表）：
┌──────────────────────────┬──────────┬──────────┬──────────┬─────────┐
│ 规则                     │ 维度     │ 阈值     │ 实际值   │ 状态    │
├──────────────────────────┼──────────┼──────────┼──────────┼─────────┤
│ order_id非空             │ 完整性   │ 100%     │ 100%     │ ✅ PASS │
│ user_id非空              │ 完整性   │ 100%     │ 99.97%   │ ⚠️ WARN │
│ order_status值域         │ 准确性   │ 95%      │ 100%     │ ✅ PASS │
│ amount > 0               │ 有效性   │ 99%      │ 99.85%   │ ✅ PASS │
│ order_id唯一             │ 唯一性   │ 100%     │ 100%     │ ✅ PASS │
│ 创建时间<当前时间        │ 有效性   │ 100%     │ 100%     │ ✅ PASS │
│ user_id存在于users表     │ 一致性   │ 99.9%    │ 99.92%   │ ✅ PASS │
│ 数据延迟<30分钟          │ 及时性   │ 95%      │ 87.3%    │ ❌ FAIL │
├──────────────────────────┼──────────┼──────────┼──────────┼─────────┤
│ 综合得分                 │          │          │ 96.2/100 │ B+      │
└──────────────────────────┴──────────┴──────────┴──────────┴─────────┘

等级标准：A(95+) / B+(90-95) / B(85-90) / C(70-85) / D(<70)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

数据血缘示例（订单报表数据流）：
┌──────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐
│ MySQL    │     │ Kafka    │     │ Flink    │     │ Hive DW  │
│ orders表 │────>│ order    │────>│ 实时计算  │────>│ dws_order│
│          │     │ _events  │     │ _summary │     │ _daily   │
└──────────┘     └──────────┘     └──────────┘     └─────┬────┘
                                                         │
                              ┌───────────────────────────┤
                              ▼                           ▼
                        ┌──────────┐              ┌──────────┐
                        │ ADS报表  │              │ BI看板   │
                        │ ads_gmv  │─────────────>│ Tableau  │
                        │ _daily   │              │ Dashboard│
                        └──────────┘              └──────────┘

 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

"""基于SQL解析的血缘采集器"""
from sqlglot import parse_one, exp
from sqlglot.lineage import lineage as sqlglot_lineage
from typing import Set, Tuple

class SQLLineageParser:
    """解析SQL语句，提取表级和列级血缘"""
    
    def extract_table_lineage(self, sql: str) -> dict:
        """
        提取表级血缘
        返回: {"source_tables": [...], "target_table": "...", "join_type": "..."}
        """
        parsed = parse_one(sql)
        
        sources = set()
        target = None
        
        # 提取源表（FROM/JOIN子句）
        for table in parsed.find_all(exp.Table):
            table_name = f"{table.db}.{table.name}" if table.db else table.name
            sources.add(table_name)
        
        # 提取目标表（INSERT INTO / CREATE TABLE AS）
        if isinstance(parsed, exp.Insert):
            target_table = parsed.find(exp.Table)
            target = f"{target_table.db}.{target_table.name}" if target_table.db else target_table.name
            # INSERT语句中目标表不应在sources中
            sources.discard(target)
        
        elif isinstance(parsed, exp.Create):
            target_table = parsed.find(exp.Table)
            target = f"{target_table.db}.{target_table.name}" if target_table.db else target_table.name
            sources.discard(target)
        
        return {
            "source_tables": list(sources),
            "target_table": target,
            "sql_type": type(parsed).__name__
        }
    
    def extract_column_lineage(self, sql: str, target_table: str) -> list:
        """
        提取列级血缘
        返回每个目标列的来源列映射
        """
        column_mappings = []
        
        for col_lineage in sqlglot_lineage(
            column="*",
            sql=sql,
            schema={}
        ):
            column_mappings.append({
                "target_column": col_lineage.column.name,
                "source_column": col_lineage.source.name,
                "source_table": col_lineage.source.table,
                "expression": str(col_lineage.expression)
            })
        
        return column_mappings


# 使用示例
parser = SQLLineageParser()

sql = """
INSERT INTO dws.dws_order_daily (order_date, user_id, total_amount, order_count)
SELECT 
    DATE(created_at) AS order_date,
    user_id,
    SUM(amount) AS total_amount,
    COUNT(*) AS order_count
FROM ods.ods_orders
WHERE created_at >= '2026-06-01'
GROUP BY DATE(created_at), user_id
"""

result = parser.extract_table_lineage(sql)
print(result)
# 输出: {
#   "source_tables": ["ods.ods_orders"],
#   "target_table": "dws.dws_order_daily",
#   "sql_type": "Insert"
# }

 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

// Neo4j 图数据库 - 血缘关系存储模型

// 创建表节点
CREATE (t:Table {
    name: 'ods.ods_orders',
    database: 'mysql_prod',
    schema: 'order_db',
    row_count: 15000000,
    last_updated: datetime('2026-06-20T02:00:00')
})

// 创建列节点
CREATE (c:Column {
    name: 'amount',
    data_type: 'DECIMAL(10,2)',
    nullable: false,
    comment: '订单金额'
})

// 创建血缘关系
CREATE (t1:Table {name: 'ods.ods_orders'})
CREATE (t2:Table {name: 'dws.dws_order_daily'})
CREATE (t1)-[:FEEDS_INTO {
    transformation: 'SUM(amount)',
    last_executed: datetime('2026-06-20T03:00:00'),
    job_name: 'daily_order_aggregation'
}]->(t2)

// 列级血缘
CREATE (c1:Column {name: 'amount', table: 'ods.ods_orders'})
CREATE (c2:Column {name: 'total_amount', table: 'dws.dws_order_daily'})
CREATE (c1)-[:DERIVES_TO {
    expression: 'SUM(amount)',
    transformation_type: 'aggregation'
}]->(c2)

// 查询：某张表的所有上游依赖
MATCH (upstream:Table)-[:FEEDS_INTO*]->(target:Table {name: 'ads.ads_gmv_daily'})
RETURN upstream.name, length(relationships(path)) as distance

// 查询：某个字段的影响范围（下游追踪）
MATCH path = (c:Column {name: 'amount', table: 'ods.ods_orders'})
       -[:DERIVES_TO*]->(downstream:Column)
RETURN downstream.table, downstream.name, length(path) as hop_count

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

数据资产目录架构：
┌──────────────────────────────────────────────────────────────┐
│                     数据资产门户（Web UI）                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │ 搜索发现  │  │ 资产浏览  │  │ 质量监控  │  │ 血缘可视化│   │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘   │
├──────────────────────────────────────────────────────────────┤
│                     目录服务层（Backend）                      │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  · 资产注册与分类     · 标签管理                       │   │
│  │  · 数据Owner管理      · 访问权限控制                   │   │
│  │  · 质量评分展示       · 使用热度统计                   │   │
│  └──────────────────────────────────────────────────────┘   │
├──────────────────────────────────────────────────────────────┤
│                     数据存储层                                 │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐       │
│  │PostgreSQL│  │ Neo4j   │  │Elastic- │  │ MinIO   │       │
│  │(目录数据)│  │(血缘图) │  │search   │  │(文件存储)│       │
│  │         │  │         │  │(全文检索)│  │         │       │
│  └─────────┘  └─────────┘  └─────────┘  └─────────┘       │
└──────────────────────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

数据资产分类体系（四级目录）：
├── 业务域（Domain）
│   ├── 交易域
│   │   ├── 主题域（Subject）
│   │   │   ├── 订单
│   │   │   │   ├── 数据集（Dataset）
│   │   │   │   │   ├── ods_orders（MySQL）
│   │   │   │   │   ├── dwd_order_detail（Hive）
│   │   │   │   │   └── order_realtime（Kafka）
│   │   │   ├── 支付
│   │   │   └── 退款
│   ├── 用户域
│   ├── 商品域
│   └── 营销域

  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

"""数据资产注册服务"""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI(title="数据资产目录服务")

class DataAsset(BaseModel):
    asset_id: str
    name: str
    description: str
    domain: str  # 业务域
    subject: str  # 主题域
    asset_type: str  # TABLE, API, FILE, STREAM
    technical_type: str  # MySQL, Hive, Kafka, REST...
    location: str  # 连接信息或路径
    owner: str  # 数据负责人
    owner_team: str
    tags: List[str]
    sensitivity_level: str  # PUBLIC, INTERNAL, CONFIDENTIAL, RESTRICTED
    quality_score: Optional[float] = None
    created_at: datetime = datetime.now()

class AssetCatalog:
    """资产目录管理"""
    
    def __init__(self, db, neo4j_driver, es_client):
        self.db = db
        self.graph = neo4j_driver
        self.search = es_client
    
    def register_asset(self, asset: DataAsset) -> str:
        """注册新数据资产"""
        # 1. 存储到PostgreSQL
        self.db.execute("""
            INSERT INTO data_assets 
            (asset_id, name, description, domain, subject, 
             asset_type, technical_type, location, owner, 
             owner_team, tags, sensitivity_level, quality_score)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (asset_id) DO UPDATE SET
            description = EXCLUDED.description,
            quality_score = EXCLUDED.quality_score,
            updated_at = NOW()
        """, (
            asset.asset_id, asset.name, asset.description,
            asset.domain, asset.subject, asset.asset_type,
            asset.technical_type, asset.location, asset.owner,
            asset.owner_team, asset.tags, asset.sensitivity_level,
            asset.quality_score
        ))
        
        # 2. 同步到Elasticsearch（全文检索）
        self.search.index(
            index="data_assets",
            id=asset.asset_id,
            document={
                "name": asset.name,
                "description": asset.description,
                "domain": asset.domain,
                "tags": asset.tags,
                "owner": asset.owner,
                "quality_score": asset.quality_score
            }
        )
        
        return asset.asset_id
    
    def search_assets(self, query: str, filters: dict = None) -> list:
        """全文搜索数据资产"""
        es_query = {
            "bool": {
                "must": [
                    {"multi_match": {
                        "query": query,
                        "fields": ["name^3", "description", "tags^2"]
                    }}
                ]
            }
        }
        
        if filters:
            filter_clauses = []
            if filters.get("domain"):
                filter_clauses.append(
                    {"term": {"domain": filters["domain"]}}
                )
            if filters.get("asset_type"):
                filter_clauses.append(
                    {"term": {"asset_type": filters["asset_type"]}}
                )
            if filter_clauses:
                es_query["bool"]["filter"] = filter_clauses
        
        results = self.search.search(
            index="data_assets",
            query=es_query,
            size=20
        )
        
        return [hit["_source"] for hit in results["hits"]["hits"]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

数据价值评估四维模型：
┌──────────────────────────────────────────────────────────────┐
│                                                              │
│  维度一：使用价值（Usage Value）                             │
│  · 数据被多少个应用/报表使用？                               │
│  · 数据被查询的频率？                                       │
│  · 数据支撑了多少业务决策？                                 │
│                                                              │
│  维度二：质量价值（Quality Value）                           │
│  · 数据质量评分                                             │
│  · 数据完整性、准确性                                       │
│  · 数据可用率                                               │
│                                                              │
│  维度三：稀缺性价值（Scarcity Value）                        │
│  · 数据是否可替代？                                         │
│  · 获取同等数据的成本                                       │
│  · 数据是否有时效性壁垒                                     │
│                                                              │
│  维度四：合规价值（Compliance Value）                        │
│  · 数据是否涉及法规要求（个人信息、金融数据等）              │
│  · 数据是否有商业授权价值                                   │
│  · 数据是否支撑审计合规                                     │
│                                                              │
└──────────────────────────────────────────────────────────────┘

  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

"""数据资产价值评估模型"""
import numpy as np
from dataclasses import dataclass
from typing import Dict

@dataclass
class ValueAssessment:
    asset_id: str
    usage_value: float      # 使用价值 (0-100)
    quality_value: float    # 质量价值 (0-100)
    scarcity_value: float   # 稀缺性价值 (0-100)
    compliance_value: float # 合规价值 (0-100)
    total_value: float      # 综合价值 (0-100)
    estimated_cost: float   # 估算成本（元/年）
    value_tier: str         # 等级: 核心/重要/一般/冗余

class DataValueAssessor:
    """数据价值评估器"""
    
    # 各维度权重（可根据行业调整）
    WEIGHTS = {
        "usage": 0.35,
        "quality": 0.25,
        "scarcity": 0.25,
        "compliance": 0.15
    }
    
    def assess(self, asset_id: str, metrics: Dict) -> ValueAssessment:
        """评估单个数据资产的价值"""
        
        # 1. 使用价值计算
        usage_value = self._calc_usage_value(metrics)
        
        # 2. 质量价值计算
        quality_value = self._calc_quality_value(metrics)
        
        # 3. 稀缺性价值计算
        scarcity_value = self._calc_scarcity_value(metrics)
        
        # 4. 合规价值计算
        compliance_value = self._calc_compliance_value(metrics)
        
        # 5. 综合评分（加权求和）
        total_value = (
            usage_value * self.WEIGHTS["usage"] +
            quality_value * self.WEIGHTS["quality"] +
            scarcity_value * self.WEIGHTS["scarcity"] +
            compliance_value * self.WEIGHTS["compliance"]
        )
        
        # 6. 估算存储+维护成本
        estimated_cost = self._estimate_cost(metrics)
        
        # 7. 确定价值等级
        tier = self._determine_tier(total_value, compliance_value)
        
        return ValueAssessment(
            asset_id=asset_id,
            usage_value=round(usage_value, 1),
            quality_value=round(quality_value, 1),
            scarcity_value=round(scarcity_value, 1),
            compliance_value=round(compliance_value, 1),
            total_value=round(total_value, 1),
            estimated_cost=estimated_cost,
            value_tier=tier
        )
    
    def _calc_usage_value(self, metrics: Dict) -> float:
        """
        使用价值 = f(下游依赖数, 查询频率, 用户覆盖数)
        """
        downstream_count = min(metrics.get("downstream_count", 0), 50)
        query_frequency = min(metrics.get("daily_query_count", 0), 10000)
        user_coverage = min(metrics.get("distinct_users_30d", 0), 100)
        
        # 归一化到0-100
        downstream_score = (downstream_count / 50) * 100
        frequency_score = (query_frequency / 10000) * 100
        user_score = (user_coverage / 100) * 100
        
        return np.average(
            [downstream_score, frequency_score, user_score],
            weights=[0.4, 0.3, 0.3]
        )
    
    def _calc_quality_value(self, metrics: Dict) -> float:
        """质量价值直接使用质量评分"""
        return metrics.get("quality_score", 0) * 100
    
    def _calc_scarcity_value(self, metrics: Dict) -> float:
        """
        稀缺性 = f(可替代性, 获取成本, 时效壁垒)
        """
        replaceable = metrics.get("is_replaceable", True)
        acquisition_cost = metrics.get("acquisition_cost_annual", 0)
        has_time_barrier = metrics.get("has_time_barrier", False)
        
        score = 50  # 基线
        if not replaceable:
            score += 30
        if acquisition_cost > 100000:
            score += min(acquisition_cost / 10000, 20)
        if has_time_barrier:
            score += 20
        
        return min(score, 100)
    
    def _calc_compliance_value(self, metrics: Dict) -> float:
        """合规价值评估"""
        sensitivity = metrics.get("sensitivity_level", "PUBLIC")
        regulatory_scope = metrics.get("regulatory_scope", [])
        has_business_license = metrics.get("has_business_license", False)
        
        sensitivity_scores = {
            "PUBLIC": 10,
            "INTERNAL": 30,
            "CONFIDENTIAL": 60,
            "RESTRICTED": 90
        }
        
        score = sensitivity_scores.get(sensitivity, 10)
        if "GDPR" in regulatory_scope or "个人信息保护法" in regulatory_scope:
            score += 20
        if has_business_license:
            score += 15
        
        return min(score, 100)
    
    def _estimate_cost(self, metrics: Dict) -> float:
        """估算年度成本（存储+计算+人力维护）"""
        storage_gb = metrics.get("storage_size_gb", 0)
        daily_compute_hours = metrics.get("daily_compute_hours", 0)
        
        # 存储成本：约0.5元/GB/月
        storage_cost = storage_gb * 0.5 * 12
        
        # 计算成本：约2元/小时
        compute_cost = daily_compute_hours * 365 * 2
        
        # 人力维护成本（基础）
        maintenance_cost = 5000  # 基础运维人力分摊
        
        return storage_cost + compute_cost + maintenance_cost
    
    def _determine_tier(self, total_value: float, 
                        compliance_value: float) -> str:
        """确定资产等级"""
        if compliance_value >= 80:
            return "核心"  # 合规驱动，无论使用价值如何都是核心
        elif total_value >= 80:
            return "核心"
        elif total_value >= 60:
            return "重要"
        elif total_value >= 40:
            return "一般"
        else:
            return "冗余"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

数据资产价值评估报告（部分）：
┌─────────────────────┬──────┬──────┬──────┬──────┬──────┬──────┬──────┐
│ 资产名称            │ 使用 │ 质量 │ 稀缺 │ 合规 │ 综合 │ 成本 │ 等级 │
├─────────────────────┼──────┼──────┼──────┼──────┼──────┼──────┼──────┤
│ ods_orders          │ 92   │ 96   │ 75   │ 70   │ 86.5 │ 8.2万│ 核心 │
│ dws_user_profile    │ 88   │ 91   │ 85   │ 90   │ 88.6 │ 4.5万│ 核心 │
│ ads_daily_report    │ 75   │ 88   │ 40   │ 30   │ 66.8 │ 2.1万│ 重要 │
│ dim_product_info    │ 65   │ 82   │ 55   │ 20   │ 60.0 │ 1.2万│ 重要 │
│ log_app_behavior    │ 45   │ 72   │ 30   │ 60   │ 49.5 │ 12万 │ 一般 │
│ tmp_debug_data_2024 │ 5    │ 45   │ 10   │ 10   │ 16.3 │ 3.8万│ 冗余 │
└─────────────────────┴──────┴──────┴──────┴──────┴──────┴──────┴──────┘

关键发现：
· log_app_behavior 存储成本高(12万/年)但价值评分一般，建议归档或采样
· tmp_debug_data_2024 为临时表遗留，建议确认后清理
· dws_user_profile 涉及个人信息（合规分90），需确保脱敏措施到位

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

推荐的推进策略（三阶段）：
┌──────────────────────────────────────────────────────────┐
│  第一阶段（3个月）：看清家底                              │
│  · 核心数据库的元数据采集                                 │
│  · 基本的数据资产目录（能看到有什么表）                   │
│  · 不追求完美，先覆盖80%的核心资产                       │
│                                                          │
│  第二阶段（6个月）：评估质量                              │
│  · 上线数据质量规则引擎                                   │
│  · 核心表的质量评分和监控                                 │
│  · 血缘关系追踪（先从ETL SQL解析开始）                   │
│                                                          │
│  第三阶段（12个月）：量化价值                             │
│  · 数据价值评估模型上线                                   │
│  · 资产目录全面开放给业务方                               │
│  · 数据成本分摊和ROI分析                                  │
└──────────────────────────────────────────────────────────┘