feat: 完成 issue #29 TDengine 超级表 schema 依据点位字典自动生成
This commit is contained in:
@@ -10,7 +10,7 @@ topic / 时序库表 / 关系表 / 对象桶：换行业只改模板资产（模
|
||||
| 文件 | 职责 |
|
||||
|------|------|
|
||||
| `templating.py` | 模板命名推导：Kafka topic + 分区策略、TDengine 超级表/子表、PostgreSQL schema/表、MinIO 桶/对象键 |
|
||||
| `tdengine_schema.py` | 时序库 schema 自动生成（超级表 + 每测点子表）+ 批量 INSERT SQL |
|
||||
| `tdengine_schema.py` | 时序库 schema 自动生成（超级表 + 每测点子表，**依据点位字典**）+ 批量 INSERT SQL |
|
||||
| `postgres_schema.py` | 关系库 schema（模板 / 模型 / 用户 / 权限）+ 角色授权语句 |
|
||||
| `batch_writer.py` | 批量写入缓冲：批量聚合（默认 5000 条/0.1s，对齐 PRD 5.2「5k/100ms」基线）、幂等去重、失败重试 —— **数据不丢不重** |
|
||||
|
||||
@@ -18,9 +18,15 @@ topic / 时序库表 / 关系表 / 对象桶：换行业只改模板资产（模
|
||||
|
||||
```python
|
||||
from data_bus.batch_writer import BatchWriter, TdengineSink
|
||||
from data_bus.tdengine_schema import generate_schema, specs_from_point_dict_rows
|
||||
from data_bus.templating import TemplateNaming
|
||||
|
||||
naming = TemplateNaming(template="ti-cl4") # 换行业只改模板名
|
||||
|
||||
# 依据点位字典自动生成 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)
|
||||
|
||||
sink = TdengineSink(naming, executor=run_sql) # executor 注入 TDengine 连接适配器
|
||||
writer = BatchWriter(sink, batch_size=5000, flush_interval=0.1)
|
||||
|
||||
|
||||
@@ -28,8 +28,10 @@ from .postgres_schema import generate_grant_ddl, generate_schema_ddl
|
||||
from .tdengine_schema import (
|
||||
PointSpec,
|
||||
build_batch_insert,
|
||||
generate_schema,
|
||||
generate_subtable_ddls,
|
||||
generate_supertable_ddl,
|
||||
specs_from_point_dict_rows,
|
||||
)
|
||||
from .templating import TemplateNaming, sanitize, sanitize_sql
|
||||
|
||||
@@ -40,6 +42,8 @@ __all__ = [
|
||||
"PointSpec",
|
||||
"generate_supertable_ddl",
|
||||
"generate_subtable_ddls",
|
||||
"generate_schema",
|
||||
"specs_from_point_dict_rows",
|
||||
"build_batch_insert",
|
||||
"generate_schema_ddl",
|
||||
"generate_grant_ddl",
|
||||
|
||||
@@ -17,6 +17,7 @@ spec.loader.exec_module(pkg)
|
||||
|
||||
from data_bus import BatchWriter, MemorySink, TdengineSink, TemplateNaming, __version__ # noqa: E402
|
||||
from data_bus.postgres_schema import generate_schema_ddl # noqa: E402
|
||||
from data_bus.tdengine_schema import generate_schema, specs_from_point_dict_rows # noqa: E402
|
||||
|
||||
ddl = generate_schema_ddl(TemplateNaming(template="ti-cl4"))
|
||||
assert "DEFAULT '{}'::jsonb" in ddl, "jsonb default escaped wrongly"
|
||||
@@ -24,3 +25,14 @@ assert "REFERENCES tpl_ti_cl4.templates(template)" in ddl
|
||||
print("version:", __version__)
|
||||
print("ddl ok; lines:", len(ddl.splitlines()))
|
||||
print("exports ok:", [c.__name__ for c in (BatchWriter, MemorySink, TdengineSink)])
|
||||
|
||||
# 子任务 #29：依据点位字典自动生成 TDengine 完整 schema（超级表 + 子表）
|
||||
_td_rows = [
|
||||
{"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "unit": "℃", "dataType": "float"},
|
||||
{"device_id": "CLF-02", "point_id": "CLF-02.RUN", "unit": "%", "dataType": "bool"},
|
||||
]
|
||||
_td_ddls = generate_schema(TemplateNaming(template="ti-cl4"), specs_from_point_dict_rows(_td_rows), retention_days=90)
|
||||
assert _td_ddls[0].startswith("CREATE STABLE IF NOT EXISTS ti_cl4_points")
|
||||
assert len(_td_ddls) == 3 # 1 超级表 + 2 子表
|
||||
assert all("USING ti_cl4_points" in d for d in _td_ddls[1:])
|
||||
print("tdengine schema ok; statements:", len(_td_ddls))
|
||||
|
||||
@@ -120,6 +120,49 @@ def generate_subtable_ddls(
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 依据点位字典自动生成完整 schema（子任务 #29 交付入口）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def specs_from_point_dict_rows(rows: List[dict]) -> List[PointSpec]:
|
||||
"""把点位字典 CSV 行（edge-gateway 点位字典表头）转换为 PointSpec 列表。
|
||||
|
||||
与 `core/edge-gateway/point_dict/loader.py` 的 CSV 表头对齐：
|
||||
``device_id / point_id / unit / dataType``（兼容小写 ``data_type``）。
|
||||
其余列（name / sampleRate / qualityCode / opcNode / protocol）与
|
||||
schema 生成无关，宽松忽略；非法行按 ``from_dict`` 宽松兜底，
|
||||
不做静默丢弃（保持与点位数一致，便于外层校验定位）。
|
||||
|
||||
Args:
|
||||
rows: 点位字典行（每行一个测点，dict 键为 CSV 表头）。
|
||||
|
||||
Returns:
|
||||
与输入顺序一致的 PointSpec 列表。
|
||||
"""
|
||||
return [PointSpec.from_dict(raw) for raw in rows]
|
||||
|
||||
|
||||
def generate_schema(
|
||||
naming: TemplateNaming,
|
||||
points: List[PointSpec],
|
||||
retention_days: Optional[int] = None,
|
||||
) -> List[str]:
|
||||
"""依据点位字典一次性生成完整 TDengine schema（超级表 + 每测点子表）。
|
||||
|
||||
Args:
|
||||
naming: 模板命名器（决定超级表/子表名）。
|
||||
points: 点位字典（测点维度集合，可由 specs_from_point_dict_rows 得到）。
|
||||
retention_days: 数据保留天数（KEEP，可选）。
|
||||
|
||||
Returns:
|
||||
DDL 语句列表：第 0 条为超级表 CREATE STABLE，其余为子表
|
||||
CREATE TABLE ... USING ... TAGS(...)，可直接按序执行。
|
||||
"""
|
||||
return [generate_supertable_ddl(naming, retention_days=retention_days)] + generate_subtable_ddls(
|
||||
naming, points
|
||||
)
|
||||
|
||||
|
||||
def build_batch_insert(
|
||||
naming: TemplateNaming,
|
||||
rows: List[dict],
|
||||
|
||||
@@ -10,8 +10,10 @@ import _bootstrap # noqa: F401
|
||||
from data_bus.tdengine_schema import (
|
||||
PointSpec,
|
||||
build_batch_insert,
|
||||
generate_schema,
|
||||
generate_subtable_ddls,
|
||||
generate_supertable_ddl,
|
||||
specs_from_point_dict_rows,
|
||||
)
|
||||
from data_bus.templating import TemplateNaming
|
||||
|
||||
@@ -22,6 +24,19 @@ def sample_point(device_id="CLF-01", point_id="CLF-01.TEMP", unit="℃", data_ty
|
||||
return PointSpec(device_id=device_id, point_id=point_id, unit=unit, data_type=data_type)
|
||||
|
||||
|
||||
# 与 edge-gateway 点位字典 CSV 表头对齐的样例行
|
||||
SAMPLE_POINT_DICT_ROWS = [
|
||||
{"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "name": "炉温", "unit": "℃",
|
||||
"dataType": "float", "sampleRate": "1000", "qualityCode": "true", "opcNode": "ns=2;s=CLF.Temp"},
|
||||
{"device_id": "CLF-01", "point_id": "CLF-01.PRES", "name": "炉压", "unit": "kPa",
|
||||
"dataType": "float", "sampleRate": "1000", "qualityCode": "true"},
|
||||
{"device_id": "CLF-02", "point_id": "CLF-02.RUN", "name": "运行状态", "unit": "%",
|
||||
"dataType": "bool", "sampleRate": "1000"},
|
||||
{"device_id": "S7-01", "point_id": "S7-01.PUMP_A", "name": "泵A频率", "unit": "Hz",
|
||||
"dataType": "float", "sampleRate": "1000", "protocol": "s7"},
|
||||
]
|
||||
|
||||
|
||||
class SupertableDDLTest(unittest.TestCase):
|
||||
def test_ddl_shape(self):
|
||||
ddl = generate_supertable_ddl(NAMING)
|
||||
@@ -72,5 +87,49 @@ class BatchInsertTest(unittest.TestCase):
|
||||
self.assertEqual(build_batch_insert(NAMING, []), [])
|
||||
|
||||
|
||||
class PointDictSchemaTest(unittest.TestCase):
|
||||
"""依据点位字典自动生成完整 schema（子任务 #29 交付口径）。"""
|
||||
|
||||
def test_specs_from_point_dict_rows(self):
|
||||
specs = specs_from_point_dict_rows(SAMPLE_POINT_DICT_ROWS)
|
||||
self.assertEqual(len(specs), 4)
|
||||
self.assertEqual(specs[0].device_id, "CLF-01")
|
||||
self.assertEqual(specs[0].point_id, "CLF-01.TEMP")
|
||||
self.assertEqual(specs[0].unit, "℃")
|
||||
self.assertEqual(specs[0].data_type, "float")
|
||||
# 兼容小写 data_type 键
|
||||
specs2 = specs_from_point_dict_rows(
|
||||
[{"device_id": "CLF-01", "point_id": "P1", "unit": "℃", "data_type": "int"}]
|
||||
)
|
||||
self.assertEqual(specs2[0].data_type, "int")
|
||||
# 缺失列宽松兜底
|
||||
specs3 = specs_from_point_dict_rows([{"device_id": "D1", "point_id": "P2"}])
|
||||
self.assertEqual(specs3[0].unit, "")
|
||||
self.assertEqual(specs3[0].data_type, "float")
|
||||
|
||||
def test_generate_schema_full(self):
|
||||
specs = specs_from_point_dict_rows(SAMPLE_POINT_DICT_ROWS)
|
||||
ddls = generate_schema(NAMING, specs)
|
||||
# 第 0 条是超级表，其余是子表；4 个测点 → 1 + 4 条
|
||||
self.assertEqual(len(ddls), 5)
|
||||
self.assertTrue(ddls[0].startswith("CREATE STABLE IF NOT EXISTS ti_cl4_points"))
|
||||
self.assertTrue(all(d.startswith("CREATE TABLE IF NOT EXISTS ti_cl4_pt_") for d in ddls[1:]))
|
||||
# 子表均 USING 同一超级表，且含点位字典标签
|
||||
self.assertIn("USING ti_cl4_points", ddls[1])
|
||||
self.assertIn("'CLF-01'", ddls[1])
|
||||
# bool 点位标签保留 data_type
|
||||
self.assertTrue(any("'bool'" in d for d in ddls))
|
||||
|
||||
def test_generate_schema_retention(self):
|
||||
specs = specs_from_point_dict_rows(SAMPLE_POINT_DICT_ROWS)
|
||||
ddls = generate_schema(NAMING, specs, retention_days=365)
|
||||
self.assertIn("KEEP(365)", ddls[0])
|
||||
|
||||
def test_generate_schema_empty_points(self):
|
||||
ddls = generate_schema(NAMING, [])
|
||||
self.assertEqual(len(ddls), 1) # 只有超级表，无子表
|
||||
self.assertIn("CREATE STABLE", ddls[0])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user