8 - 电商问数：表与字段信息同步到元数据库

本章课程目标：

  • 理解为什么元数据知识库的第一步，一定是先把表信息和字段信息沉淀到 Meta MySQL。
  • 看懂“配置描述 -> 业务实体 -> DW 补齐真实信息 -> Meta MySQL 落库”这条完整链路。
  • 建立清晰的分层认识：Service 负责组织流程，DW Repository 负责查询真实结构，Meta Repository 负责落库，Mapper 负责对象转换。

学习建议： 这一章先抓住一条落库链路：配置文件描述表字段，系统去数仓补齐真实结构，再转换成内部实体，最后写入 Meta MySQL。ORM、Repository、事务这些细节都服务于这条链路。读代码时重点看哪些数据来自配置，哪些数据来自 DW，哪些数据最终沉淀到元数据库。

对应代码分支： 08-metadata-table-column-sync

上一章我们已经把“元数据知识库构建”的入口说明白了：脚本如何启动、配置如何加载、服务层如何接过 config_path 并进入真正的业务流程。

从这一章开始，我们正式进入第一条真正落地的业务链路：把配置文件里描述的表和字段，转换成系统内部统一的数据对象，并写入 Meta MySQL。

这一步很基础，但也非常关键。因为后面的字段向量索引、字段取值全文索引、SQL 生成上下文组织，全部都建立在这批结构化元数据之上。也就是说，只有表信息和字段信息先稳定落下来，后面的检索和问数链路才有可靠的基础。

1、本章在整条构建流程中的位置

回到 MetaKnowledgeService.build(config_path)，表链路对应的是下面这一段：

项目对应文件路径：shopkeeper-agent/app/services/meta_knowledge_service.py

1
2
3
4
5
6
7
8
9
async def build(self, config_path: Path):
    context = OmegaConf.load(config_path)
    schema = OmegaConf.structured(MetaConfig)
    meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

    if meta_config.tables:
        column_infos = await self._save_tables_to_meta_db(meta_config)
        await self._save_column_info_to_qdrant(column_infos)
        await self._save_value_info_to_es(meta_config, column_infos)

只要配置里存在 tables，系统就会依次做 3 件事：

  1. 先把表信息和字段信息写入 Meta MySQL
  2. 再根据 column_infos 为字段建立向量索引
  3. 最后根据配置中的 sync 标记，为部分字段建立字段值全文索引

本章只讲第一步，也就是：

1
column_infos = await self._save_tables_to_meta_db(meta_config)

可以把这一章看作：先把“字段是什么”这件事讲清楚，后面再讲“字段怎么被检索”。

2、先抓主线：_save_tables_to_meta_db(…) 到底做了什么

进入细节之前，先看本章最核心的方法。

项目对应文件路径：shopkeeper-agent/app/services/meta_knowledge_service.py

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
async def _save_tables_to_meta_db(self, meta_config: MetaConfig) -> list[ColumnInfo]:
    table_infos: list[TableInfo] = []
    column_infos: list[ColumnInfo] = []

    for table in meta_config.tables:
        table_info = TableInfo(
            id=table.name,
            name=table.name,
            role=table.role,
            description=table.description,
        )
        table_infos.append(table_info)

        column_types: dict[str, str] = await self.dw_mysql_repository.get_column_types(
            table.name
        )
        for column in table.columns:
            column_values: list = await self.dw_mysql_repository.get_column_values(
                table.name, column.name, 10
            )
            column_info = ColumnInfo(
                id=f"{table.name}.{column.name}",
                name=column.name,
                type=column_types[column.name],
                role=column.role,
                examples=column_values,
                description=column.description,
                alias=column.alias,
                table_id=table.name,
            )
            column_infos.append(column_info)

    async with self.meta_mysql_repository.session.begin():
        await self.meta_mysql_repository.save_table_infos(table_infos)
        await self.meta_mysql_repository.save_column_infos(column_infos)

    return column_infos

它做的事情其实非常清楚：

  1. 遍历配置中的每张表，先构造 TableInfo
  2. 去 DW 查询这张表所有字段的真实类型
  3. 再遍历这张表下的每个字段，查询一部分真实取值作为示例
  4. 构造 ColumnInfo
  5. 开启事务，把表信息和字段信息统一写入 Meta MySQL

所以后面整章的内容，本质上就是把这 5 步一层层拆开来看。

3、第一步：配置里的表和字段，如何变成业务实体

这一层最重要的认知是：配置文件不是直接入库的，配置文件会先被转换成业务实体。

如果你对“配置文件、业务实体、ORM 模型、mappers”这几类角色的区别还不够稳，这里建议先回看第 7 章前面的概念说明。第 8 章从这里开始，默认你已经先接受一个前提：

  • 配置文件负责描述“这次要处理什么”
  • 业务实体负责表示“系统内部怎么统一表达这些数据”
  • Mapper 负责在业务实体和 ORM 模型之间做转换
  • ORM 模型负责把数据映射到数据库表

也就是说，这一章不是把 YAML 逐行塞进数据库，而是先把配置中的描述组织成系统内部统一的数据对象。

3.1 先看配置长什么样

项目对应文件路径：shopkeeper-agent/conf/meta_config.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
tables:
  - name: dim_region
    role: dim
    description: 地区维度表，用于描述订单发生的地理区域信息。
    columns:
      - name: region_id
        role: primary_key
        description: 地区唯一标识。
        alias: [地区ID, 区域ID]
        sync: false

      - name: region_name
        role: dimension
        description: 订单所属的大区名称，如华东、华南等。
        alias: [地区, 区域, 大区]
        sync: true

这份配置主要负责告诉系统两件事：哪些表、哪些字段要进入元数据知识库；这些表和字段的业务语义是什么。

这里能看到的内容，大多是业务描述信息，比如：表角色 role，字段角色 role，字段说明

3.2 再看系统内部统一的数据对象

项目对应文件路径：

  • shopkeeper-agent/app/entities/table_info.py
  • shopkeeper-agent/app/entities/column_info.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@dataclass
class TableInfo:
    id: str
    name: str
    role: str
    description: str


@dataclass
class ColumnInfo:
    id: str
    name: str
    type: str
    role: str
    examples: list[Any]
    description: str
    alias: list[str]
    table_id: str

这里的 TableInfo 是系统内部流转的表元数据对象，ColumnInfo 则是系统内部流转的字段元数据对象。

它们还不是 ORM 模型，而是更轻量、更偏业务语义的数据载体。

3.3 为什么 ColumnInfo.id 不是直接用字段名

对应代码：

1
id=f"{table.name}.{column.name}"

原因很简单：真实数仓里会有大量同名字段。例如很多表里都会出现：id、customer_id、product_id、date_id等同名字段。

如果只用字段名做唯一标识，不同表中的同名字段很快就会冲突。所以当前项目把字段 ID 设计成：表名 + 点号 + 字段名

例如：

  • fact_order.order_amount
  • dim_customer.customer_id
  • dim_product.category

这样字段在系统内部才具备稳定、明确的唯一性。

3.4 为什么 type 和 examples 不是直接从配置里来

看 ColumnInfo 的构造代码：

1
2
3
4
5
6
7
8
9
10
column_info = ColumnInfo(
    id=f"{table.name}.{column.name}",
    name=column.name,
    type=column_types[column.name],
    role=column.role,
    examples=column_values,
    description=column.description,
    alias=column.alias,
    table_id=table.name,
)

这里很值得注意：

  • role、description、alias 来自配置
  • type、examples 来自运行时查询

这背后对应的是一个很重要的工程原则：能从系统自动获取的真实信息，就尽量不要让用户手工重复维护。

字段类型本来就是数据库表结构的一部分，示例值本来就存在于真实数据里。如果这些内容也要求用户在配置里手填，会带来两个明显问题：维护成本高，很容易和真实数仓脱节。所以配置文件主要负责描述业务语义，而程序运行时负责补齐真实结构和真实样本。

4、第二步：为什么要专门有一个 DW Repository

前面已经看到，字段类型和字段示例值都不是配置里直接给出的，而是运行时从数仓中查出来的。
这部分职责，就落在 DWMySQLRepository 上。

项目对应文件路径：shopkeeper-agent/app/repositories/mysql/dw/dw_mysql_repository.py

1
2
3
4
5
6
7
8
9
10
async def get_column_types(self, table_name: str) -> dict[str, str]:
    sql = f"show columns from {table_name}"
    result = await self.session.execute(text(sql))
    return {row.Field: row.Type for row in result.fetchall()}


async def get_column_values(self, table_name: str, column_name: str, limit: int):
    sql = f"select distinct {column_name} from {table_name} limit {limit}"
    result = await self.session.execute(text(sql))
    return result.scalars().fetchall()

这两个方法分别做两件事：

  • get_column_types(table_name)：查一张表所有字段的真实类型
  • get_column_values(table_name, column_name, limit)：查某个字段的一部分真实取值

4.1 为什么字段类型是一张表查一次

先看这行代码：

1
column_types = await self.dw_mysql_repository.get_column_types(table.name)

它放在外层表循环里，是因为字段类型的查询粒度是“按表”。

例如查询 fact_order，返回结果可能类似：

1
2
3
4
5
{
    "order_id": "varchar(30)",
    "customer_id": "varchar(20)",
    "order_amount": "float",
}

这样后面遍历字段时，就可以直接通过：

1
column_types[column.name]

拿到当前字段的真实类型。

4.2 为什么字段示例值要一列一查

再看这行代码：

1
2
3
column_values = await self.dw_mysql_repository.get_column_values(
    table.name, column.name, 10
)

它放在内层字段循环里，是因为示例值的查询粒度是“按字段”。

例如：

1
select distinct category from dim_product limit 10

可能返回：

1
["饮料", "休闲食品", "粮油调味", "家居清洁"]

这些值最终会写入：

1
examples=column_values

写进 ColumnInfo.examples。

这里要特别注意一点：这一章拿到的是“示例值”，不是该字段的全部取值。

所以这一层的目标不是为了直接构建全文索引，而是为了让字段元数据先带上一小部分有代表性的真实样本。真正的大规模字段值同步，会在后面的 Elasticsearch 章节展开。

5、第三步：为什么 Service 先处理业务实体，而不是直接操作 ORM

这是本章非常值得建立起来的分层意识。当前项目里，Service 层并没有直接构造 SQLAlchemy 的 ORM 模型，而是先处理：TableInfo、ColumnInfo这样的业务实体。这样做的关键目的，不是“多套一层”，而是为了解耦。

如果 Service 层直接依赖 ORM 模型，会带来一个明显问题：业务逻辑会直接绑定到底层存储实现，一旦 ORM 模型变化，业务层代码也要跟着改。

而现在这样分层之后：

  • Service 层只关心业务对象怎么组织
  • Repository 层才关心底层如何写入数据库

这也是为什么后面同一批 ColumnInfo 还能继续被拿去构建 Qdrant 向量索引、参与 Elasticsearch 取值索引，而不必让业务层到处带着 ORM 对象跑。

6、第四步：Meta Repository 和 Mapper 到底负责什么

现在我们已经有了两类业务实体：TableInfo、ColumnInfo。接下来要做的，就是把它们写入 Meta MySQL。这部分职责在 MetaMySQLRepository 中。

项目对应文件路径：shopkeeper-agent/app/repositories/mysql/meta/meta_mysql_repository.py

1
2
3
4
5
6
7
8
async def save_table_infos(self, table_infos: list[TableInfo]):
    models = [TableInfoMapper.to_model(table_info) for table_info in table_infos]
    self.session.add_all(models)


async def save_column_infos(self, columns_info: list[ColumnInfo]):
    models = [ColumnInfoMapper.to_model(column_info) for column_info in columns_info]
    self.session.add_all(models)

这里最关键的点，不是 add_all()，而是：Repository 对外暴露的输入仍然是业务实体，而不是 ORM 模型。

也就是说，Service 层只需要把“业务上已经整理好的数据对象”交给仓储层，至于怎么转换、怎么落库，由仓储层继续往下吸收。

6.1 Mapper 的作用是翻译

项目对应文件路径：

  • shopkeeper-agent/app/repositories/mysql/meta/mappers/table_info_mapper.py
  • shopkeeper-agent/app/repositories/mysql/meta/mappers/column_info_mapper.py
1
2
3
4
5
6
7
8
9
10
class TableInfoMapper:
    @staticmethod
    def to_model(table_info: TableInfo) -> TableInfoMySQL:
        return TableInfoMySQL(**asdict(table_info))


class ColumnInfoMapper:
    @staticmethod
    def to_model(column_info: ColumnInfo) -> ColumnInfoMySQL:
        return ColumnInfoMySQL(**asdict(column_info))

这里的角色分工非常清楚：

  • TableInfo、ColumnInfo 是业务实体
  • TableInfoMySQL、ColumnInfoMySQL 是 ORM 模型
  • Mapper 负责把前者转换成后者

所以整条写库链路可以压缩成下面这样：

1
2
3
4
5
MetaConfig
  -> TableInfo / ColumnInfo
  -> TableInfoMapper / ColumnInfoMapper
  -> TableInfoMySQL / ColumnInfoMySQL
  -> Meta MySQL

6.2 add_all() 不等于真正提交数据库

很多初学者第一次接触 ORM 时，看到：

1
self.session.add_all(models)

很容易以为“已经写进数据库了”。其实不是。

add_all() 做的事情，本质上只是：

  • 把这些 ORM 对象加入当前 Session
  • 告诉 ORM：这些对象后续需要被持久化

它本身还不是最终的提交动作。真正的数据落库，要发生在事务提交阶段。

7、第五步：为什么事务要放在 Service 层统一管理

回到本章主方法：

1
2
3
async with self.meta_mysql_repository.session.begin():
    await self.meta_mysql_repository.save_table_infos(table_infos)
    await self.meta_mysql_repository.save_column_infos(column_infos)

这里使用的是：

1
async with session.begin():

这种写法在当前场景下非常合适，因为它把事务边界说得很清楚：“保存表信息”和“保存字段信息”属于同一批业务操作，它们要么一起成功，要么一起失败。

这背后体现的是一个很重要的分层原则：

  • Repository 负责“怎么写”
  • Service 负责“这些写操作要以什么业务事务组织起来”

所以事务边界放在 Service 层，是合理的。它还有两个很实用的好处：1. 如果上下文中的操作都成功，事务会自动提交。2. 如果中途抛出异常，事务会自动回滚。

这样代码会比手写 try/except + commit + rollback 更简洁，也更不容易出错。

8、整条链路回顾

到这里，我们就可以把“一次表信息与字段信息同步到元数据库”的完整顺序串起来了：

  1. 脚本读取 meta_config.yaml
  2. MetaKnowledgeService.build(config_path) 进入表链路
  3. 遍历配置中的每张表，构造 TableInfo
  4. 通过 DWMySQLRepository.get_column_types(...) 查询该表所有字段的真实类型
  5. 遍历该表中的每个字段，再通过 DWMySQLRepository.get_column_values(...) 查询示例值
  6. 构造 ColumnInfo
  7. 打开一笔事务
  8. MetaMySQLRepository 接收业务实体，并通过 Mapper 转成 ORM 模型
  9. session.add_all() 把这些 ORM 对象加入当前 Session
  10. 事务结束后统一提交，数据真正写入 Meta MySQL

如果把这段流程再压缩成一句话，可以记成：配置描述先变成业务实体，业务实体再补齐真实结构与样本，最后以事务方式落到元数据库。

9、验证本章效果

这一章完成后，最合适的验证方式不是先去看 Qdrant 或 Elasticsearch，而是先确认 Meta MySQL 里的结构化元数据是否已经正确落下来。

9.1 运行同步脚本

1
python -m app.scripts.build_meta_knowledge -c conf/meta_config.yaml

9.2 重点检查 table_info 和 column_info

这一章最值得重点看的，不是有没有“插入成功”这 4 个字，而是下面这些字段是否真的带上了应有的信息。

先看 table_info：

  • 是否已经写入每张配置中的表
  • id、name、role、description 是否正确

再看 column_info：

  • id 是否是 表名.字段名
  • type 是否已经带上真实字段类型
  • examples 是否已经带上一部分真实示例值
  • description、alias、table_id 是否和配置保持一致

其中最关键的是 type 和 examples。
只要这两个字段已经有值，就说明本章最核心的“配置 + 动态补齐 + 落库”链路已经跑通了。

验证本章效果

本章小结：

  • 元数据知识库的第一步，不是建向量索引，而是先把表和字段沉淀成结构化元数据。
  • 配置文件负责描述业务语义，真实字段类型和字段示例值由程序运行时从 DW 自动补齐。
  • Service 负责编排流程，DW Repository 负责查真实结构，Meta Repository 负责落库，Mapper 负责对象转换，事务边界由 Service 统一管理。

下一章开始，我们继续沿着这条表链路往后走，进入第二步：为字段信息建立向量索引，并写入 Qdrant。