2026-08-04 16:58:06 +08:00
|
|
|
|
# Data Bus —— 数据总线 + 时序库模板化封装
|
|
|
|
|
|
|
|
|
|
|
|
对应 PRD 5.2「② 数据总线 + 时序库」与 Issue #4（EPIC）。
|
|
|
|
|
|
复用 **Kafka + TDengine + PostgreSQL + MinIO**，全部改为**按模板配置**
|
|
|
|
|
|
topic / 时序库表 / 关系表 / 对象桶：换行业只改模板资产（模板 YAML + 点位字典 CSV），
|
|
|
|
|
|
内核代码零改动。
|
|
|
|
|
|
|
|
|
|
|
|
## 模块
|
|
|
|
|
|
|
|
|
|
|
|
| 文件 | 职责 |
|
|
|
|
|
|
|------|------|
|
|
|
|
|
|
| `templating.py` | 模板命名推导：Kafka topic + 分区策略、TDengine 超级表/子表、PostgreSQL schema/表、MinIO 桶/对象键 |
|
2026-08-04 18:14:37 +08:00
|
|
|
|
| `tdengine_schema.py` | 时序库 schema 自动生成（超级表 + 每测点子表，**依据点位字典**）+ 批量 INSERT SQL |
|
2026-08-04 16:58:06 +08:00
|
|
|
|
| `postgres_schema.py` | 关系库 schema（模板 / 模型 / 用户 / 权限）+ 角色授权语句 |
|
|
|
|
|
|
| `batch_writer.py` | 批量写入缓冲：批量聚合（默认 5000 条/0.1s，对齐 PRD 5.2「5k/100ms」基线）、幂等去重、失败重试 —— **数据不丢不重** |
|
|
|
|
|
|
|
|
|
|
|
|
## 使用示例
|
|
|
|
|
|
|
|
|
|
|
|
```python
|
|
|
|
|
|
from data_bus.batch_writer import BatchWriter, TdengineSink
|
2026-08-04 18:14:37 +08:00
|
|
|
|
from data_bus.tdengine_schema import generate_schema, specs_from_point_dict_rows
|
2026-08-04 16:58:06 +08:00
|
|
|
|
from data_bus.templating import TemplateNaming
|
|
|
|
|
|
|
|
|
|
|
|
naming = TemplateNaming(template="ti-cl4") # 换行业只改模板名
|
2026-08-04 18:14:37 +08:00
|
|
|
|
|
|
|
|
|
|
# 依据点位字典自动生成 TDengine 完整 schema（超级表 + 每测点子表）
|
|
|
|
|
|
point_dict_rows = [{"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "unit": "℃", "dataType": "float"}]
|
|
|
|
|
|
schema_ddls = generate_schema(naming, specs_from_point_dict_rows(point_dict_rows), retention_days=90)
|
|
|
|
|
|
|
2026-08-04 16:58:06 +08:00
|
|
|
|
sink = TdengineSink(naming, executor=run_sql) # executor 注入 TDengine 连接适配器
|
|
|
|
|
|
writer = BatchWriter(sink, batch_size=5000, flush_interval=0.1)
|
|
|
|
|
|
|
|
|
|
|
|
writer.push_many(rows) # 批量入队（重复键自动过滤）
|
|
|
|
|
|
writer.flush() # 主动刷盘；失败保留缓冲重发
|
|
|
|
|
|
```
|
|
|
|
|
|
|
|
|
|
|
|
## 验收口径（Issue #4）
|
|
|
|
|
|
|
|
|
|
|
|
- **端到端写入**：`push → flush → sink` 全链路（MemorySink 本地联调 / TdengineSink 落库）。
|
|
|
|
|
|
- **批量写入**：batch_size / flush_interval 触发批量 flush（默认 5000 条 / 0.1s）。
|
|
|
|
|
|
- **数据不丢不重**：写成功才清缓冲（不丢）；(device_id, point_id, ts) 幂等去重
|
|
|
|
|
|
（不重，含失败重发的部分写入场景，sink 侧二次兜底）。
|
|
|
|
|
|
|
|
|
|
|
|
## 测试
|
|
|
|
|
|
|
|
|
|
|
|
```bash
|
|
|
|
|
|
cd core/data-bus
|
|
|
|
|
|
python -m unittest discover -s tests -v
|
|
|
|
|
|
```
|
