让大模型读懂你的数据：基于 LLM 与查询日志自动构建数据资产目录（Data Catalog）

在中大型企业的数据仓库与数据湖中，随着业务快速迭代，表数量通常以每年数千张的速度膨胀。随之而来的是严重的“数据荒漠化”与暗数据（Dark Data）泛滥：

• 超过 60% 的数据表没有注释，或者注释仅有“临时表”、“test”等毫无意义的废话；
• 原作者离职后，下游数十个报表依赖的表变成了“无人敢动、无人敢删”的黑盒；
• 新来的业务分析师想找“用户近 30 天退款明细”，在数仓里搜出十几个名为 dwd_order_refund_* 的表，只能一个个在飞书群里打听。

传统 Data Catalog（如 DataHub、OpenMetadata、Atlas）主要依赖工程师手工录入元数据。但“要求开发写文档”向来是违背人性的，静态文档录入往往撑不过半年就会彻底荒废。

引入大语言模型（LLM）结合数仓查询审计日志（Audit Query Logs）与列级血缘（Column Lineage），正在彻底颠覆这一现状：系统能自动分析谁在查这张表、怎么查的、字段之间如何计算，从而自动生成高准确度的数据资产画像、业务标签与 PII 敏感合规标记。

本文深入剖析基于 LLM 的现代数据资产目录自动构建体系与生产级落地实践。

一、破局之道：为什么不能仅靠 DDL，还要读查询日志？

如果仅把 CREATE TABLE 语句丢给大模型，模型只能望文生义地“瞎猜”：例如看到字段 is_special，模型根本不知道它到底代表“特殊优惠商户”还是“内部测试账号”。

要让大模型真正理解数据资产，必须提供三重上下文融合：

+-----------------------------------------------------------------------------------+
| 1. 静态 DDL 与约束画像 (Static Schema & Profiling)                                |
|    - 字段名、物理类型、唯一值基数、空值率、Min/Max 极值                          |
+-----------------------------------------------------------------------------------+
                                          |
+-----------------------------------------------------------------------------------+
| 2. 动态 SQL 审计日志 (Dynamic Query Audit Logs) [最关键!]                         |
|    - 过去 30 天针对该表的真实查询 SQL (包含哪些 WHERE 过滤、哪些 GROUP BY 维度)  |
|    - 示例: "WHERE is_special = 1 AND is_test = 0" -> 证明 is_special 是业务特权标记|
+-----------------------------------------------------------------------------------+
                                          |
+-----------------------------------------------------------------------------------+
| 3. 上下游列级血缘 (Column-level Lineage)                                          |
|    - 上游来自哪个业务库表的哪个字段，下游被哪些高管报表消费                       |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
|                            LLM 资产理解与元数据合成                               |
|   产出: 表级业务简介 + 字段级标准化注释 + PII 敏感分级 + 典型消费场景与弃用建议    |
+-----------------------------------------------------------------------------------+

二、端到端自动化架构：从样本脱敏到 Catalog 闭环

一条生产级的数据资产目录流水线必须将数据隐私安全与自动化审核置于首位：

+------------------+      +-------------------+      +----------------------+
| 数仓元数据 & 日志 | ---> | PII 敏感样本脱敏  | ---> | LLM 语义与标签生成   |
| (Spark/Trino/ODS)|      | (正则掩码+类型化) |      | (结构化 JSON Schema) |
+------------------+      +-------------------+      +----------------------+
                                                                |
                                                                v
+------------------+      +-------------------+      +----------------------+
| OpenMetadata     | <--- | 人机协同与置信度  | <--- | 自动校验与冲突拦截   |
| / DataHub 资产库 |      | (一键确认 / 撤销) |      | (校验字段名与血缘)   |
+------------------+      +-------------------+      +----------------------+

关键工程防线：

1. 样本前置脱敏（PII Sanitization）：严禁把包含真实手机号、身份证、真实姓名的样本喂给 LLM。所有采样数据在离开本地网络前，必须经过正则掩码处理（如 138****1234）。
2. AI 标识与版本回滚（AI-Generated Tags）：所有由大模型生成的注释必须被打上 origin: AI_GENERATED 标签，在前端展示时带有明显标识，并支持数据负责人一键审核（Approve）或回滚为历史人工版本。

三、生产级数据资产分析与 OpenMetadata 注册器实现

下面的 Python 代码实现了一套工业级的资产画像生成流水线。它包含 PII 数据掩码器、结合 SQL 日志的 Prompt 组装器、结构化 JSON 输出校验，以及对 DataHub / OpenMetadata 标准格式的映射。

"""
auto_catalog_builder.py
生产级大模型驱动的数据资产目录构建与 OpenMetadata 自动注册引擎
"""

import json
import re
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class ColumnInputMeta:
    name: str
    data_type: str
    raw_comment: str = ""
    sample_values: List[str] = field(default_factory=list)


@dataclass
class TableEnrichedContext:
    table_name: str
    database: str
    columns: List[ColumnInputMeta]
    recent_query_sqls: List[str] = field(default_factory=list)
    upstream_tables: List[str] = field(default_factory=list)


@dataclass
class GeneratedColumnDoc:
    name: str
    description: str
    pii_category: Optional[str] = None  # 如: PHONE, ID_CARD, EMAIL, NONE
    business_term: Optional[str] = None


@dataclass
class GeneratedTableAsset:
    table_name: str
    table_summary: str
    domain: str                         # 主题域: 交易/用户/供应链等
    typical_use_cases: List[str]
    column_docs: Dict[str, GeneratedColumnDoc]
    confidence_score: float


class PIISanitizer:
    """PII 敏感数据清洗脱敏器，确保样本进入大模型前绝对安全"""

    PHONE_PATTERN = re.compile(r"1[3-9]\d{9}")
    ID_CARD_PATTERN = re.compile(r"\d{17}[\dXx]")
    EMAIL_PATTERN = re.compile(r"[\w\.-]+@[\w\.-]+\.\w+")

    @classmethod
    def sanitize(cls, val: str) -> str:
        if not val or not isinstance(val, str):
            return str(val)
        val = cls.PHONE_PATTERN.sub("138****0000", val)
        val = cls.ID_CARD_PATTERN.sub("110101********0000", val)
        val = cls.EMAIL_PATTERN.sub("user****@masked.com", val)
        return val


class LLMDataCatalogEngine:
    """数据资产生成与注册引擎"""

    def __init__(self, confidence_threshold: float = 0.8):
        self.confidence_threshold = confidence_threshold

    def build_prompt(self, ctx: TableEnrichedContext) -> str:
        """组装包含 DDL、脱敏样本与 SQL 审计日志的综合上下文"""
        cols_payload = []
        for col in ctx.columns:
            masked_samples = [PIISanitizer.sanitize(s) for s in col.sample_values[:3]]
            cols_payload.append(
                f"- `{col.name}` ({col.data_type}) 原注释: '{col.raw_comment}' 样本: {masked_samples}"
            )
        cols_text = "\n".join(cols_payload)

        sqls_text = "\n".join([f"  * {sql[:200]}..." for sql in ctx.recent_query_sqls[:3]])

        prompt = f"""
你是一位企业级数据治理与元数据专家。请根据以下表的元数据、脱敏样本及最近真实查询日志，分析该数据资产的业务含义并输出资产目录文档。

【数据库/表名】: {ctx.database}.{ctx.table_name}
【上游血缘表】: {', '.join(ctx.upstream_tables) if ctx.upstream_tables else '贴源接入'}
【字段列表】:
{cols_text}

【最近业务查询 SQL 日志】:
{sqls_text if sqls_text else '  暂无高频查询日志'}

【输出要求】:
严格输出标准 JSON 格式，不得包含任何 Markdown 解释：
{{
    "table_summary": "清晰描述这张表的业务定位与包含的核心数据 (2~3句话)",
    "domain": "所属业务域 (如: 交易履约 / 用户增长 / 财务结算)",
    "typical_use_cases": ["用途1: 用于计算每日实付GMV", "用途2: 用于统计各省份退款率"],
    "columns": [
        {{
            "name": "字段名",
            "description": "通俗易懂的业务中文解释",
            "pii_category": "NONE 或 PHONE/ID_CARD/BANK_CARD/NAME",
            "business_term": "绑定的企业标准词条"
        }}
    ],
    "confidence_score": 0.95
}}
"""
        return prompt

    def parse_llm_response(self, table_name: str, raw_json: str) -> GeneratedTableAsset:
        """解析大模型 JSON 输出，并进行结构化防错转换"""
        data = json.loads(raw_json)
        col_docs = {}
        for c in data.get("columns", []):
            col_name = c.get("name")
            col_docs[col_name] = GeneratedColumnDoc(
                name=col_name,
                description=c.get("description", "暂无描述"),
                pii_category=c.get("pii_category", "NONE"),
                business_term=c.get("business_term")
            )

        return GeneratedTableAsset(
            table_name=table_name,
            table_summary=data.get("table_summary", "待补充业务描述"),
            domain=data.get("domain", "通用"),
            typical_use_cases=data.get("typical_use_cases", []),
            column_docs=col_docs,
            confidence_score=float(data.get("confidence_score", 0.0))
        )

生产环境生成与注册演练

# 1. 准备富上下文输入 (包含带有业务线索的真实查询 SQL)
mock_context = TableEnrichedContext(
    database="dwd",
    table_name="dwd_trade_order_details_df",
    columns=[
        ColumnInputMeta("order_id", "string", "主键", ["ORD_20260824001", "ORD_20260824002"]),
        ColumnInputMeta("buyer_mobile", "string", "", ["13812345678", "13987654321"]), # 敏感字段
        ColumnInputMeta("real_pay_fee", "decimal(18,2)", "金额", ["99.50", "199.00"]),
        ColumnInputMeta("is_special", "int", "", ["0", "1"]),                          # 模糊字段
    ],
    recent_query_sqls=[
        "SELECT SUM(real_pay_fee) FROM dwd_trade_order_details_df WHERE is_special = 1 GROUP BY dt;",
        "SELECT buyer_mobile, count(*) FROM dwd_trade_order_details_df WHERE real_pay_fee > 1000 GROUP BY 1;"
    ],
    upstream_tables=["ods.mysql_mall_orders", "ods.mysql_mall_order_items"]
)

# 2. 模拟大模型根据 SQL 日志与上下文生成的标准 JSON
mock_llm_json_output = json.dumps({
    "table_summary": "电商零售交易明细事实表，按订单行记录买家实付金额、优惠抵扣及特殊商户标识，是计算大盘 GMV 与复购率的核心底层事实表。",
    "domain": "电商交易域",
    "typical_use_cases": [
        "每日大盘实付成交额 (GMV) 汇总与统计",
        "大促期间高净值用户消费频次与复购分析"
    ],
    "columns": [
        {"name": "order_id", "description": "全局唯一的电商子订单编号", "pii_category": "NONE", "business_term": "订单号"},
        {"name": "buyer_mobile", "description": "下单买家手机号码 (高敏感)", "pii_category": "PHONE", "business_term": "买家联系电话"},
        {"name": "real_pay_fee", "description": "剔除优惠券抵扣后的实际现金支付金额", "pii_category": "NONE", "business_term": "实付金额"},
        {"name": "is_special", "description": "特殊渠道标识位 (1 代表直播带货特价单，0 代表普通商城单)", "pii_category": "NONE", "business_term": "特殊渠道标"}
    ],
    "confidence_score": 0.92
}, ensure_ascii=False)

# 3. 解析并注册入库
engine = LLMDataCatalogEngine()
asset_doc = engine.parse_llm_response(mock_context.table_name, mock_llm_json_output)

print(f"=== 成功为表 [{asset_doc.table_name}] 生成资产目录 ===")
print(f"【业务领域】: {asset_doc.domain}")
print(f"【表级概述】: {asset_doc.table_summary}")
print(f"【置信度得分】: {asset_doc.confidence_score}")
print("【字段画像明细】:")
for col_name, doc in asset_doc.column_docs.items():
    pii_flag = f" [🚨 PII: {doc.pii_category}]" if doc.pii_category != "NONE" else ""
    print(f"  * `{col_name}`: {doc.description}{pii_flag}")

四、治理权衡与落地防线

在推进数据资产目录自动化时，必须牢记以下三条实战治理原则：

1. 自动识别僵尸表并推动下线：
   资产目录不仅要建文档，还要定期统计每张表的血缘下游与查询热度。当大模型结合审计日志发现某张表“近 90 天被查询次数为 0，且无任何下游生产依赖”时，自动标记为“疑似僵尸资产（Zombie Asset）”，并给原负责人推送下线确认工单，节约数仓存储与计算资源。
2. 结合语义搜索打造自然语言寻表助手：
   将生成的表描述、字段注释向量化存入向量库。业务同学无需知道物理表名，直接在 Data Catalog 搜索框输入：“我想找能看每个城市退款率的表”，系统通过语义检索直接定位到 dwd_trade_order_details_df 并给出推荐 SQL 示例。
3. 把控责任归属（Ownership）底线：
   大模型可以帮我们理解字段语义，但表的业务负责人（Owner）与数据安全等级审批人必须严格来自于组织架构或研发工单流，绝不能交由模型猜测。

通过“元数据 + 查询日志 + LLM 语义推理 + 严格样本脱敏”的闭环，数据团队能够以极低的人力成本将数千张沉睡的黑盒数据表激活为可检索、可理解、可治理的高价值资产。