大模型辅助数仓维度建模：从 ODS 逆向工程到一致性总线矩阵设计

在数据仓库与数据湖构建中，最考验数据架构师功力的环节往往是维度建模（Dimensional Modeling）。

无论是经典的 Kimball 星型模型，还是现代湖仓一体下的宽表与语义层架构，将上游错综复杂的数十张业务 OLTP 表清洗、重构为高内聚、低耦合的 DWD/DWS 数据资产，通常需要耗费数周的调研与评审：

• 业务过程（Business Process）如何界定？
• 事实表的粒度（Grain）如何做到“最小不可分割”？
• 哪些字段应该沉淀为独立维度，哪些字段应当退化（Degenerate Dimension）？
• 如何保证不同主题域共用一致性维度（Conformed Dimensions），避免“同名不同义”或“同义不同名”？

引入大语言模型（LLM）并不是让 AI 代替架构师去拍板业务定义，而是利用大模型强大的语义推理与元数据理解能力，充当“建模副驾驶（Modeling Copilot）”：自动化完成 ODS 实体逆向工程、事实/维度字段初筛、总线矩阵冲突检查以及 DDL/dbt 骨架生成。

本文深入剖析基于 LLM 的数仓维度建模辅助体系与生产级落地实践。

---

一、AI 辅助建模全景：Kimball 方法论的现代工程化

基于大模型的维度建模必须严格锚定经典的 Kimball 四步建模法，将其解构为标准化的自动化与人机协同环节：

+-----------------------------------------------------------------------------------+
| 1. 业务过程识别 (Select the Business Process)                                     |
|    - 输入: 业务需求文档 (PRD)、OLTP 库 DDL、业务变更日志 (CDC)                     |
|    - LLM 动作: 识别核心业务事件 (下单、支付、履约、退款、评价)                    |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 2. 声明粒度 (Declare the Grain)                                                   |
|    - LLM 动作: 结合主键与业务事件，明确事实表每行的物理意义 (如"每个子订单行")   |
|    - 防御机制: 严禁混合粒度建模 (如严禁在一张表中同时记录订单级与商品行级汇总)   |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 3. 维度识别与一致性对齐 (Identify the Dimensions)                                 |
|    - LLM 动作: 结合字段画像 (Cardinality/类型/分布) 划分实体维表与退化维          |
|    - 总线矩阵校验: 对齐企业级一致性维度 (如公共用户维表 dim_user、公共地理维表)  |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 4. 确定事实与度量 (Identify the Facts)                                            |
|    - LLM 动作: 提取可加性度量 (Additive)、半可加度量与事务快照状态               |
|    - 产物输出: 输出标准 DDL 建表语句、dbt 模型脚手架与 Data Catalog 元数据文档    |
+-----------------------------------------------------------------------------------+

---

二、智能建模流水线架构

一个健壮的 AI 建模流水线绝不能仅靠一句简单的 Prompt。它必须通过特征抽取、总线矩阵校验与架构师闭环裁决来保证设计质量：

+------------------+      +-------------------+      +----------------------+
|  ODS 贴源层元数据 | ---> | 字段与统计画像提炼 | ---> | LLM 语义与关系抽取   |
| (DDL/注释/样本)  |      | (基数/分布/外键)  |      | (业务过程/维度/事实)  |
+------------------+      +-------------------+      +----------------------+
                                                                |
                                                                v
+------------------+      +-------------------+      +----------------------+
| 产出 DWD/DWS DDL | <--- | 人工裁决与架构终审 | <--- | 一致性总线矩阵冲突检查|
| & 数据资产目录   |      | (消解歧义/确立口径) |      | (跨域维度一致性对比) |
+------------------+      +-------------------+      +----------------------+

关键防线：

1. 基数与分布驱动：纯文本大模型无法感知数据真实分布。流水线必须先运行轻量 Profiler 采集字段基数（Distinct Count）、空值率与数值极差，防止模型把高基数订单号误划分为枚举维度。
2. 总线矩阵（Bus Matrix）冲突检查：当模型建议新建一个 dim_shop_v2 时，系统自动比对数仓现有的一致性总线矩阵。如果已存在 dim_merchant，立即提示冲突并强制对齐，杜绝烟囱式建表。

---

三、生产级 AI 建模分析器与 DDL 生成器实现

下面的 Python 实现展示了一套完整的智能建模辅助引擎。它包含字段特征解析、业务术语注入、结构化星型模型推断以及一致性总线矩阵检查。

"""
dim_modeling_copilot.py
生产级大模型辅助数仓维度建模与总线矩阵校验引擎
"""

import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Set


class DimensionType(Enum):
    CONFORMED = "CONFORMED"      # 一致性公共维度
    DEGENERATE = "DEGENERATE"    # 退化维度 (直接存放在事实表)
    JUNK = "JUNK"                # 杂项/标志位维度
    SCD2 = "SCD2"                # 缓慢变化维 (拉链表)


@dataclass
class ColumnProfile:
    name: str
    data_type: str
    cardinality: int             # 唯一值基数
    null_ratio: float
    comment: str = ""
    sample_values: List[Any] = field(default_factory=list)


@dataclass
class StarSchemaDesign:
    """星型模型设计方案"""
    table_name: str
    business_process: str
    grain: str
    dimension_columns: List[str]
    degenerate_dimensions: List[str]
    fact_metrics: List[str]
    foreign_keys: Dict[str, str]  # 字段名 -> 关联的公共维表
    rationale: str


class BusMatrixCatalog:
    """企业级一致性总线矩阵库"""
    def __init__(self):
        # 预置数仓已有的一致性公共维度与基准表
        self.conformed_dimensions: Dict[str, str] = {
            "user_id": "dim_user_global",
            "buyer_id": "dim_user_global",
            "merchant_id": "dim_merchant",
            "shop_id": "dim_merchant",
            "sku_id": "dim_goods_sku",
            "spu_id": "dim_goods_spu",
            "city_id": "dim_geo_region",
        }

    def check_dimension_alignment(self, proposed_fks: Dict[str, str]) -> List[str]:
        """检查建议关联的外键维表是否符合一致性总线规范"""
        warnings = []
        for col, target_dim in proposed_fks.items():
            if col in self.conformed_dimensions:
                standard_dim = self.conformed_dimensions[col]
                if target_dim != standard_dim:
                    warnings.append(
                        f"⚠️ 维度冲突: 字段 '{col}' 建议关联 '{target_dim}'，但企业总线标准为 '{standard_dim}'"
                    )
        return warnings


class AIModelingEngine:
    """维度建模核心引擎"""

    def __init__(self, bus_matrix: BusMatrixCatalog, enterprise_glossary: Dict[str, str]):
        self.bus_matrix = bus_matrix
        self.glossary = enterprise_glossary

    def build_prompt_context(self, table_name: str, columns: List[ColumnProfile]) -> str:
        """构建富含字段画像与企业术语的高密度提示词"""
        cols_summary = []
        for c in columns:
            cols_summary.append(
                f"- 字段: `{c.name}` ({c.data_type}), 注释: '{c.comment}', 基数: {c.cardinality}, 样本: {c.sample_values[:3]}"
            )
        cols_text = "\n".join(cols_summary)
        glossary_text = json.dumps(self.glossary, ensure_ascii=False)

        prompt = f"""
你是一位资深数仓架构专家。请根据以下 ODS 贴源表特征与企业术语表，设计规范的 DWD 维度建模方案。

【源表名称】: {table_name}
【企业标准术语表】: {glossary_text}
【字段画像】:
{cols_text}

【设计要求】:
1. 明确业务过程与不可分割的物理粒度 (Grain)。
2. 识别度量字段 (Facts)、退化维度 (Degenerate Dims) 与需要外挂关联的维度字段。
3. 严格输出标准 JSON 结构，格式如下：
{{
    "business_process": "业务过程描述",
    "grain": "粒度描述 (例如: 每个订单明细行一条记录)",
    "fact_metrics": ["pay_amount", "discount_amount"],
    "degenerate_dimensions": ["order_sn", "tracking_no"],
    "dimension_columns": ["buyer_id", "merchant_id", "status"],
    "foreign_keys": {{"buyer_id": "dim_user_global", "merchant_id": "dim_merchant"}},
    "rationale": "设计权衡说明"
}}
"""
        return prompt

    def generate_dwd_ddl(self, design: StarSchemaDesign, columns_dict: Dict[str, ColumnProfile]) -> str:
        """根据审核通过的模型设计方案自动编译出生产级 DDL 语句"""
        fields_ddl = []
        
        # 1. 维度与退化维
        for dim in design.degenerate_dimensions + design.dimension_columns:
            dtype = columns_dict[dim].data_type if dim in columns_dict else "STRING"
            cmt = columns_dict[dim].comment if dim in columns_dict else "维度字段"
            fields_ddl.append(f"    {dim:<24} {dtype:<16} COMMENT '{cmt}'")

        # 2. 事实度量
        for fact in design.fact_metrics:
            dtype = columns_dict[fact].data_type if fact in columns_dict else "DECIMAL(18,2)"
            cmt = columns_dict[fact].comment if fact in columns_dict else "度量指标"
            fields_ddl.append(f"    {fact:<24} {dtype:<16} COMMENT '{cmt}'")

        cols_body = ",\n".join(fields_ddl)
        
        ddl = f"""-- ====================================================================
-- 表名: dwd_{design.table_name}
-- 业务过程: {design.business_process}
-- 数据粒度: {design.grain}
-- 生成机制: AI Modeling Copilot 辅助生成，经架构师终审
-- ====================================================================
CREATE TABLE IF NOT EXISTS dwd_{design.table_name} (
{cols_body}
)
USING iceberg
PARTITIONED BY (days(created_at))
TBLPROPERTIES (
    'write.format.default' = 'parquet',
    'history.expire.max-snapshot-age-ms' = '604800000'
);
"""
        return ddl

生产建模流程演练与总线检查

# 1. 准备源表字段画像
mock_profiles = [
    ColumnProfile("order_id", "BIGINT", cardinality=1000000, null_ratio=0.0, comment="主键ID"),
    ColumnProfile("order_sn", "STRING", cardinality=1000000, null_ratio=0.0, comment="对外订单编号", sample_values=["SN20260824001"]),
    ColumnProfile("buyer_id", "BIGINT", cardinality=50000, null_ratio=0.0, comment="买家用户ID"),
    ColumnProfile("merchant_id", "BIGINT", cardinality=1200, null_ratio=0.0, comment="店铺商户ID"),
    ColumnProfile("pay_amount", "DECIMAL(18,2)", cardinality=45000, null_ratio=0.0, comment="实付金额"),
    ColumnProfile("order_status", "STRING", cardinality=6, null_ratio=0.0, comment="订单状态", sample_values=["PAID", "SHIPPED"]),
    ColumnProfile("created_at", "TIMESTAMP", cardinality=980000, null_ratio=0.0, comment="创建时间戳"),
]
col_map = {c.name: c for c in mock_profiles}

# 2. 模拟大模型输出的建模建议 (故意包含一个偏离企业总线规范的维表建议)
mock_llm_json = {
    "business_process": "电商线上零售交易下单与支付",
    "grain": "每个主订单一条记录",
    "fact_metrics": ["pay_amount"],
    "degenerate_dimensions": ["order_sn", "order_status"],
    "dimension_columns": ["buyer_id", "merchant_id"],
    "foreign_keys": {
        "buyer_id": "dim_user_global",
        "merchant_id": "dim_shop_local_v1"  # 偏离了标准 dim_merchant
    },
    "rationale": "order_sn 为高基数退化维；buyer 与 merchant 关联公共维度表"
}

# 3. 运行总线矩阵冲突检查
bus_catalog = BusMatrixCatalog()
conflict_warnings = bus_catalog.check_dimension_alignment(mock_llm_json["foreign_keys"])
print("【总线矩阵冲突审查】:")
for w in conflict_warnings:
    print(w)

# 4. 架构师纠正后，编译生成生产级 DDL
design_obj = StarSchemaDesign(
    table_name="trade_orders_df",
    business_process=mock_llm_json["business_process"],
    grain=mock_llm_json["grain"],
    dimension_columns=mock_llm_json["dimension_columns"],
    degenerate_dimensions=mock_llm_json["degenerate_dimensions"],
    fact_metrics=mock_llm_json["fact_metrics"],
    foreign_keys={"buyer_id": "dim_user_global", "merchant_id": "dim_merchant"}, # 架构师纠偏
    rationale=mock_llm_json["rationale"]
)

engine = AIModelingEngine(bus_catalog, {"实付金额": "扣除优惠券后的用户实际扣款"})
final_ddl = engine.generate_dwd_ddl(design_obj, col_map)
print("\n【生成的生产级 Iceberg DDL】:\n" + final_ddl)

---

四、能力边界与架构师的不可替代性

在将大模型引入数仓建模的实践中，必须清晰界定“AI 能做什么，不能做什么”：

1. 复杂业务口径与财税合规不能靠 AI 脑补

例如涉及跨境电商的“增值税（VAT）计提”、“多币种汇率结算点”等高风险财务口径，大模型无法通过字面猜测准确。必须由业务领域专家（SME）提供结构化业务术语表（Glossary），并由架构师完成口径复核。

2. 缓慢变化维（SCD2）与拉链表的设计权衡

模型倾向于将所有带有更新特征的字段一律推荐为拉链表（SCD Type 2）。但在千万级日增量的事实场景下，盲目建拉链表会导致存储和计算膨胀。架构师必须结合数仓 SLA、计算引擎开销和下游使用习惯，在 SCD Type 1（覆盖写）、SCD Type 2（拉链历史）与日快照（Daily Snapshot）之间做出工程权衡。

3. 一致性总线矩阵的全局治理

大模型通常是以“单表或单个主题”为窗口进行推理，容易缺乏跨越“供应链”、“营销”、“财务”、“用户增长”的全域全局视角。因此，维护全局总线矩阵、确保一致性维度的唯一命名与主键生命周期，是架构师最核心的护城河。

将繁琐的元数据解析、字段归类和 DDL 样板代码交给 AI，把业务理解、架构权衡与数据资产全局治理留给工程师，这才是数仓架构进化的最优解。