06 | 把 meta_config 同步进 MySQL（生成阶段）

06 | 把 meta_config 同步进 MySQL（生成阶段）需要先看清适用场景和关键步骤，避免只记结论却忽略实际限制。

06 | 把 meta_config 同步进 MySQL（生成阶段）

这是一篇系列文，请按顺序阅读。

本文目标

NL2SQL 召回前，要先把业务元数据灌进各个存储：

1. MySQL meta 库：结构化元数据（表 / 字段 / 指标 / 关联）
2. Qdrant：字段语义向量（后续）
3. Elasticsearch：维度值全文检索（后续）

本文完成第 1 步：把 conf/meta_config.yaml 同步到 MySQL meta 库的四张表。

数据源头是 YAML，不是直接扫业务库；业务库（dw）只用来补字段类型和样例值。

1. 整体流程

meta_config.yaml │ ▼conf/sync_db.py │ ├─ sync_table_info() ├─ sync_column_info() ← 会读 dw 库补 type / examples ├─ sync_metric_info() └─ sync_column_metric()│▼ Service → Repository → Mapper → Model│▼ MySQL meta 库

调用链一句话：

2. meta 库要存什么

关系：

table_info 1 ── N column_infocolumn_info N ── N metric_info （通过 column_metric）

配置上做了拆分：

1. tables：只放业务表（dim / fact）及其 columns
2. metric_info：单独放在顶级 key，不再塞进 tables

这样同步逻辑更清晰，也避免把指标误写进 table_info。

3. 为什么要分层

YAML 格式和数据库表结构不一样，不能「读完 YAML 直接 INSERT」。分层是为了把职责拆开：

相关文件：

app/├── entities/# table_info / column_info / metric_info / column_metric├── models/├── mappers/├── repositories/# 含 dw_db_repository（读业务库）├── services/└── dbs/ # meta_db.py / dw_db.pyconf/├── meta_config.yaml├── app_config.yaml├── get_config.py├── sync_db.py└── ../.env

4. 配置：密码放 .env，YAML 只放占位符

4.1 .env

MYSQL_USER=yanMYSQL_PASSWORD=你的密码LLM_API_KEY=...LLM_BASE_URL=...

4.2 app_config.yaml（片段）

meta_db:host:localhostport:3306user:"${MYSQL_USER}"password:"${MYSQL_PASSWORD}"database:metadw_db:host:localhostport:3306user:"${MYSQL_USER}"password:"${MYSQL_PASSWORD}"database:dw

4.3 get_config.py 做什么

1. load_dotenv(.env) 把环境变量读进进程
2. 读 YAML 时，把 ${MYSQL_PASSWORD} 替换成真实值
3. 缺少环境变量时直接报错，避免连库时才发现密码为空

依赖：

uv add pyyaml python-dotenv sqlalchemy asyncmy

5. 先实现 table_info：按层搭起来

后面三张表的 Entity / Model / Mapper / Repository / Service 都照这个套路搭，本文只详细展开 table_info。

5.1 Entity

# app/entities/table_info.pyfrom dataclasses import dataclass@dataclassclassTableInfo:id: strname: strrole: strdescription: str

要点：

1. id 是 str，对应数据库 varchar(64) 主键
2. YAML 没有单独 id，同步时用物理表名（如 dim_region）当稳定主键

5.2 Model

# app/models/base.pyfrom sqlalchemy.orm import DeclarativeBaseclassBase(DeclarativeBase):pass

# app/models/table_info_model.pyclassTableInfoModel(Base):__tablename__ = "table_info"id: Mapped[str] = mapped_column(String(64), primary_key=True)name: Mapped[str | None] = mapped_column(String(128))role: Mapped[str | None] = mapped_column(String(32))description: Mapped[str | None] = mapped_column(Text)

Base 统一持有 metadata，后面的 create_all / drop_all 都靠它。

5.3 Mapper

classTableInfoMapper:defto_entity(self, model: TableInfoModel) -> TableInfo:# ORM → Entity：字段铺开写（ORM.__dict__ 含内部状态）return TableInfo(id=model.id,name=model.name,role=model.role,description=model.description,)defto_model(self, entity: TableInfo) -> TableInfoModel:# Entity → ORM：dataclass.__dict__ 干净，可解包return TableInfoModel(entity.__dict__)

5.4 数据库连接

app/dbs/meta_db.py 负责 meta 库；app/dbs/dw_db.py 同理，只是读 app_config["dw_db"]。

要点：

1. 引擎模块级创建一次
2. 驱动用 mysql+asyncmy
3. 密码含特殊字符时要 quote_plus

5.5 Repository / Service

Repository 只做持久化；Service 目前是薄封装，YAML → Entity 的解析先放在 sync_db.py。

add_all 是真正的批量：一次 add_all + 一次 commit。不要在 Service 里循环调单条 add。

6. 同步脚本骨架

conf/sync_db.py 的职责：

1. 开发阶段：drop_all + create_all，方便反复测试
2. 依次同步四张表
3. 打开 Session，装配 Repository → Service，批量写入

asyncdefreset_meta_tables() ->None:asyncwith async_engine.begin() as conn:await conn.run_sync(Base.metadata.drop_all)await conn.run_sync(Base.metadata.create_all)asyncdefsync_table_info(config: dict) ->list[TableInfo]:table_infos = []for table in config.get("tables", []):name = table["name"]table_infos.append(TableInfo(id=name,name=name,role=table["role"],description=table["description"],))asyncwith MetaAsyncSessionLocal() as session:service = TableInfoService(TableInfoRepository(session))returnawait service.add_all(table_infos)

开发期用 drop_all，是为了避免重复执行时报主键冲突。代价是每次都会清空已有元数据；上线后应改成 upsert。

指标配置单独放在顶级：

metric_info:-name:GMVdescription:全称GrossMerchandiseValue，表示所有订单的成交金额总和。alias: [成交总额, 订单总额]relevant_columns:-fact_order.order_amount

7. 执行与验证

7.1 确认 MySQL 已启动

docker ps --filter name=mysql# 期望看到 Up ... (healthy) 且 0.0.0.0:3306->3306/tcp

7.2 执行同步

uv run python conf/sync_db.py

四张表都写完后，输出大致如下：

已写入 table_info 5 条已写入 column_info 24 条已写入 metric_info 2 条已写入 column_metric 2 条

7.3 查库确认

set -a && source .env && set +adocker exec -e MYSQL_PWD="$MYSQL_PASSWORD" mysql mysql --default-character-set=utf8mb4 -u"$MYSQL_USER" -D meta -e "SELECT * FROM table_info;"

--default-character-set=utf8mb4 要加，否则中文会变成 ?????。

8. 常见问题

9. 同步 column_info

YAML 里字段长这样：

columns:-name:provincerole:dimensiondescription:订单所属的省份名称。alias: [省份, 省, 所在省份]sync:true

落库字段：

id / name / type / role / description / alias / examples / table_id

映射规则：

1. id：{table_name}.{column_name}，例如 dim_region.province
2. name / role / description / alias：直接来自 YAML
3. table_id：所属业务表名
4. type：从 dw 库的 information_schema 读取
5. examples：当 sync: true 时，从 dw 表 SELECT DISTINCT 取值；否则为空列表

所以要先补业务库只读能力：

# app/repositories/dw_db_repository.pyclassDwDBRepository:asyncdefget_column_type(self, table_name: str, column_name: str) ->str | None:# SELECT COLUMN_TYPE FROM information_schema.COLUMNS ...asyncdefget_column_all_values(self, table_name: str, column_name: str) ->list[str]:# SELECT DISTINCT `column` FROM `table` WHERE `column` IS NOT NULL

Service 只是薄封装。同步核心逻辑：

asyncdefsync_column_info(config: dict) ->list[ColumnInfo]:column_infos: list[ColumnInfo] = []asyncwith DwAsyncSessionLocal() as dw_session:dw_service = DwDBService(DwDBRepository(dw_session))for table in config.get("tables", []):table_name = table["name"]for column in table.get("columns", []):column_name = column["name"]column_type = await dw_service.get_column_type(table_name, column_name)examples = (await dw_service.get_column_all_values(table_name, column_name)if column.get("sync", False)else [])column_infos.append(ColumnInfo(id=f"{table_name}.{column_name}",name=column_name,type=column_type,role=column["role"],examples=examples,description=column["description"],alias=column.get("alias", []),table_id=table_name,))asyncwith MetaAsyncSessionLocal() as meta_session:service = ColumnInfoService(ColumnInfoRepository(meta_session))returnawait service.add_all(column_infos)

Entity / Model / Mapper / Repository / Service 的搭法与 table_info 相同。

10. 同步 metric_info

YAML：

metric_info:-name:GMVdescription:全称GrossMerchandiseValue，表示所有订单的成交金额总和。alias: [成交总额, 订单总额]relevant_columns:-fact_order.order_amount

落库字段：

id / name / description / alias / relevant_columns

其中 id 直接用 name，其余字段原样写入。

asyncdefsync_metric_info(config: dict) ->list[MetricInfo]:metric_infos: list[MetricInfo] = []for record in config.get("metric_info", []):metric_infos.append(MetricInfo(id=record["name"],name=record["name"],description=record["description"],alias=record.get("alias", []),relevant_columns=record.get("relevant_columns", []),))asyncwith MetaAsyncSessionLocal() as session:service = MetricInfoService(MetricInfoRepository(session))returnawait service.add_all(metric_infos)

执行后应看到：

已写入 metric_info 2 条:- GMV- AOV

11. 同步 column_metric

column_metric 是字段与指标的多对多关联表，方便后续按字段反查可参与的指标。

它不单独出现在 YAML 里，而是从 metric_info[].relevant_columns 展开：

column_id = relevant_column # 如 fact_order.order_amountmetric_id = record["name"]# 如 GMV

因为 column_info.id 也用「表名.字段名」，所以这里可以直接对齐。

asyncdefsync_column_metric(config: dict) ->list[ColumnMetric]:column_metrics: list[ColumnMetric] = []for record in config.get("metric_info", []):for relevant_column in record.get("relevant_columns", []):column_metrics.append(ColumnMetric(column_id=relevant_column,metric_id=record["name"],))asyncwith MetaAsyncSessionLocal() as session:service = ColumnMetricService(ColumnMetricRepository(session))returnawait service.add_all(column_metrics)

执行后应看到：

已写入 column_metric 2 条:- fact_order.order_amount -> GMV- fact_order.order_quantity -> AOV

联表验证：

docker exec -e MYSQL_PWD="$MYSQL_PASSWORD" mysql mysql --default-character-set=utf8mb4 -u"$MYSQL_USER" -D meta -e "SELECT cm.column_id, cm.metric_idFROM column_metric cm;"

12. 下一步

MySQL meta 四张表已经齐了。后续继续在 sync_db.py 里补：

1. Qdrant：把字段 / 指标的描述与别名向量化，供语义召回
2. Elasticsearch：把 sync: true 的维度值写入全文索引，供值匹配

入口里已经预留：

sync_to_qdrant(meta_config)# TODOsync_to_elasticsearch(meta_config) # TODO