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

import pymysql
from sqlalchemy import create_engine, inspect

def collect_mysql_metadata(connection_string):
    """采集 MySQL 数据库的技术元数据"""
    engine = create_engine(connection_string)
    inspector = inspect(engine)

    metadata = {
        "database": engine.url.database,
        "tables": []
    }

    for table_name in inspector.get_table_names():
        table_meta = {
            "name": table_name,
            "columns": [],
            "indexes": [],
            "primary_key": inspector.get_pk_constraint(table_name),
            "foreign_keys": inspector.get_foreign_keys(table_name),
            "row_count": get_row_count(engine, table_name),
            "last_updated": get_last_updated(engine, table_name)
        }

        for column in inspector.get_columns(table_name):
            table_meta["columns"].append({
                "name": column["name"],
                "type": str(column["type"]),
                "nullable": column["nullable"],
                "default": str(column.get("default", ""))
            })

        for index in inspector.get_indexes(table_name):
            table_meta["indexes"].append({
                "name": index["name"],
                "columns": index["column_names"],
                "unique": index["unique"]
            })

        metadata["tables"].append(table_meta)

    return metadata

def get_row_count(engine, table_name):
    """获取表的行数（采样估算）"""
    with engine.connect() as conn:
        result = conn.execute(f"SELECT COUNT(*) FROM {table_name}")
        return result.scalar()

def get_last_updated(engine, table_name):
    """获取表的最近更新时间"""
    with engine.connect() as conn:
        result = conn.execute(
            f"SELECT UPDATE_TIME FROM information_schema.TABLES "
            f"WHERE TABLE_NAME = '{table_name}'"
        )
        row = result.fetchone()
        return str(row[0]) if row else None

1

元数据采集（定时任务）→ 元数据库（MySQL）→ 数据目录 API → 前端展示

1

源系统 → 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

import sqlparse
from sqlparse.sql import IdentifierList, Identifier
from sqlparse.tokens import Keyword, DML

def extract_lineage_from_sql(sql):
    """从 SQL 中提取表级血缘"""
    parsed = sqlparse.parse(sql)[0]
    sources = []
    target = None

    for token in parsed.tokens:
        if token.ttype is DML and token.value.upper() == 'INSERT':
            # 下一个 Identifier 就是目标表
            pass
        if token.ttype is Keyword and token.value.upper() == 'FROM':
            # 后面的 Identifier 就是源表
            pass
        if token.ttype is Keyword and token.value.upper() == 'INTO':
            # 后面的 Identifier 就是目标表
            pass

    return {"sources": sources, "target": target}

# 更实用的方案：用 sqllineage 库
from sqllineage.runner import LineageRunner

sql = """
INSERT INTO dwd_order_detail
SELECT o.order_id, o.amount, u.user_name
FROM ods_order o
JOIN ods_user u ON o.user_id = u.user_id
"""

runner = LineageRunner(sql)
print("源表:", runner.source_tables)   # [ods_order, ods_user]
print("目标表:", runner.target_tables)  # [dwd_order_detail]

 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

class DataQualityChecker:
    """数据质量检测引擎"""

    def __init__(self, engine):
        self.engine = engine
        self.results = []

    def check_completeness(self, table, column, threshold=0.95):
        """完整性检查：非空率是否达标"""
        sql = f"""
        SELECT COUNT(*) as total,
               COUNT({column}) as non_null
        FROM {table}
        """
        result = self.engine.execute(sql).fetchone()
        rate = result['non_null'] / result['total']
        self.results.append({
            "table": table,
            "column": column,
            "dimension": "completeness",
            "score": rate,
            "threshold": threshold,
            "passed": rate >= threshold
        })
        return rate

    def check_uniqueness(self, table, column):
        """唯一性检查：是否有重复值"""
        sql = f"""
        SELECT COUNT(*) as total,
               COUNT(DISTINCT {column}) as unique_count
        FROM {table}
        """
        result = self.engine.execute(sql).fetchone()
        rate = result['unique_count'] / result['total']
        self.results.append({
            "table": table,
            "column": column,
            "dimension": "uniqueness",
            "score": rate,
            "threshold": 1.0,
            "passed": rate == 1.0
        })
        return rate

    def check_timeliness(self, table, time_column, max_delay_hours=24):
        """及时性检查：数据是否在预期时间内更新"""
        sql = f"SELECT MAX({time_column}) as latest FROM {table}"
        result = self.engine.execute(sql).fetchone()
        latest = result['latest']
        delay = (datetime.now() - latest).total_seconds() / 3600
        self.results.append({
            "table": table,
            "dimension": "timeliness",
            "score": max(0, 1 - delay / max_delay_hours),
            "threshold": 0.8,
            "passed": delay <= max_delay_hours
        })
        return delay

    def get_quality_score(self):
        """计算综合质量分"""
        if not self.results:
            return 0
        return sum(r["score"] for r in self.results) / len(self.results)

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

数据资产健康度总览
━━━━━━━━━━━━━━━━━━━━━━━
总体质量分: 87.3 / 100 ✅

按维度：
  完整性: 92.1  ✅
  准确性: 85.4  ⚠️
  一致性: 78.6  ❌
  及时性: 95.2  ✅
  唯一性: 88.7  ✅
  合规性: 83.9  ⚠️

按数据域：
  客户域: 91.2  ✅
  订单域: 88.5  ✅
  产品域: 82.3  ⚠️
  财务域: 76.8  ❌

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

def calculate_usage_value(table_meta, access_logs):
    """计算数据的使用价值"""

    # 1. 下游依赖数（血缘分析得出）
    downstream_count = table_meta.get("downstream_count", 0)
    downstream_score = min(downstream_count / 10, 1.0) * 40  # 最多 40 分

    # 2. 访问频率（最近 30 天）
    access_count = sum(1 for log in access_logs
                       if log["table"] == table_meta["name"]
                       and log["days_ago"] <= 30)
    access_score = min(access_count / 100, 1.0) * 30  # 最多 30 分

    # 3. 使用方数量
    consumer_count = len(set(log["consumer"] for log in access_logs
                            if log["table"] == table_meta["name"]))
    consumer_score = min(consumer_count / 5, 1.0) * 30  # 最多 30 分

    return downstream_score + access_score + consumer_score

 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

def calculate_asset_value(table_meta, quality_score, usage_score):
    """计算数据资产综合价值"""

    # 质量价值（0-100）
    quality_value = quality_score * 0.3

    # 使用价值（0-100）
    usage_value = usage_score * 0.4

    # 稀缺价值（0-100）
    scarcity = table_meta.get("scarcity_score", 50)  # 人工评估
    scarcity_value = scarcity * 0.3

    total = quality_value + usage_value + scarcity_value

    # 分级
    if total >= 80:
        level = "核心资产"
    elif total >= 60:
        level = "重要资产"
    elif total >= 40:
        level = "一般资产"
    else:
        level = "低价值资产"

    return {
        "table": table_meta["name"],
        "total_score": round(total, 1),
        "level": level,
        "quality_value": round(quality_value, 1),
        "usage_value": round(usage_value, 1),
        "scarcity_value": round(scarcity_value, 1)
    }