lingniu-vehicle-ingest/docs/architecture/tdengine-batch-writer-design.md

TDengine Batch Writer Design

更新时间：2026-07-03

Problem

history-writer 当前按 Kafka 消息逐条处理，并对每个 raw frame/location 调用 TDengine 写入。这个路径简单、可恢复，但在 10,000+ FPS 后会优先暴露三个瓶颈：

• 每帧一次 SQL/网络往返，吞吐上限低。
• Kafka offset 逐条提交，commit 开销随帧率线性增长。
• burst 流量下无法用批量 flush 平滑 TDengine 写入压力。

Target

按 topic/protocol/目标表分组批量写 TDengine，按数量或时间触发 flush。

初始建议：

参数	初始值
raw frame batch size	200
location batch size	500
flush interval	100ms
operation timeout	5s
max SQL payload	8MB guardrail
Kafka commit	TDengine batch 成功后提交本批最大 offset

Write Semantics

1. Kafka consumer 拉取消息后解析一次 envelope。
2. envelope 已携带 parsed_fields 时，TDengine 写入直接复用，不重新从 raw JSON 解析。
3. raw frame 和 location 可以分批写，但 offset 只能在本消息涉及的所有写入成功后提交。
4. TDengine 写入失败时不提交 offset，让 Kafka 自动重放。
5. 依赖 event_id 保持幂等，避免失败重放造成重复业务记录。

Non-goals

• 第一阶段不改变 raw frame 表结构。
• 第一阶段不合并不同 raw frame。
• 不丢弃 parse error frame。
• 不把 MySQL snapshot 或 Redis 当前态塞进 history-writer。

Risks

• 批次越大，失败重试的重复写风险越高，必须依赖 event_id 或等价键。
• parsed_json 较大时容易让单条 SQL 超限，需要按 payload size 提前切批。
• batch flush 需要暴露 pending、flush duration、write error metrics，否则问题只会表现为 Kafka lag。

Metrics Required

指标	含义
vehicle_history_batch_flush_total{status}	批写成功/失败次数
vehicle_history_batch_rows_total{status}	批写行数
vehicle_history_batch_flush_duration_ms{status}	批写耗时
vehicle_history_kafka_lag	下游是否追得上 Kafka

Rollout

1. 保留当前逐条写实现作为 fallback。
2. 增加批写器单元测试，覆盖 count flush、interval flush、失败不 commit。
3. 在测试环境打开 batch writer。
4. 生产先用小批次 100/100ms，观察 TDengine latency 和 Kafka lag。
5. 逐步提升到 200-500/100ms。

Implementation Status

2026-07-03 已完成第一阶段实现：

• history.Writer.AppendAllBatch 按 raw child table 和 location child table 生成多行 INSERT ... VALUES (...),(...)。
• oversized payload chunks 仍写入 raw_frame_payload_chunks，同样按 child table 批量写。
• history-writer 默认启动 HISTORY_WORKERS=3 个同组消费者，每个 worker 按 HISTORY_BATCH_SIZE=200、HISTORY_BATCH_WAIT_MS=20 收集其已分配分区的消息。
• TDengine batch 成功后才批量提交 Kafka messages；batch 失败不提交 offset，让 Kafka 保留可重放语义。
• 保留 processHistoryMessage 和 AppendAll 单条路径，便于回退和测试。

新增指标：

vehicle_history_batch_flush_total{status}
vehicle_history_batch_rows_total{status}
vehicle_history_batch_flush_duration_ms{status}

生产 rollout 建议：

1. 先部署默认 200/100ms。
2. 观察 vehicle_history_batch_* 和 vehicle_history_kafka_lag。
3. 如果 TDengine latency 上升或 Kafka lag 不下降，将 HISTORY_BATCH_SIZE 降到 100。
4. 如果稳定且有 backlog，再逐步提升到 500。