从 OpenTSDB 迁到 KWDB：Schema 改造、历史数据迁移、双写切换全记录

今天想跟大家聊聊我们把一套冷链运输平台数据完整迁出的过程。车上每 30 秒上报一次温度、电池电压和车厢湿度，这些数据在 OpenTSDB 里已经存了两年，采集端和告警服务一直靠它跑。之所以要迁，一是 OpenTSDB 这套架构维护起来不轻松，底下依赖 HBase，组件多、排查链路长；二是数据一多，查询、备份和容量成本都会往上走。而采集和告警是持续在跑的，停几天迁完，业务等不起。也想验证下 KWDB 在时序场景能不能平滑接住，给后面其他系统探个路。

想讲这件事，是因为难点真不在导数据，而在线上迁移。不是简单停掉 OpenTSDB、导个文件、再导进 KWDB 就完事：导出期间数据还在往里写，告警查询也不能断，很多案例一带而过，实际最容易出问题。这里把迁移拆成了四件事，记录一套可执行的操作经验。

先把迁移链路定下来：KDTS（KaiwuDB 官方提供的迁移工具）已有 OpenTSDB 到 KWDB 的导入能力，下面只处理一个业务差异：默认按 metric 建表，而平台需要统一长表，并在切换期间保留可追踪的校验和补偿。

Schema 改造：先确定一张目标表

OpenTSDB 的点包含 metric、timestamp、value 和 tags。如果直接按 metric 自动建表，迁移后会得到多张结构相近的表。冷链平台需要统一查询温度、电压和湿度，因此把 metric 也作为标签保存：

{
  "metric": "vehicle.coolant_temp",
  "timestamp": 1719820800,
  "value": 4.6,
  "tags": {
    "vehicle_id": "TRUCK-018",
    "route_id": "R-07",
    "sensor_id": "coolant-1"
  }
}

CREATE TABLE vehicle_metrics (
    ts TIMESTAMPTZ NOT NULL,
    value DOUBLE NOT NULL
) TAGS (
    metric_name VARCHAR(64) NOT NULL,
    vehicle_id VARCHAR(32) NOT NULL,
    route_id VARCHAR(32) NOT NULL,
    sensor_id VARCHAR(32) NOT NULL
) PRIMARY TAGS (metric_name, vehicle_id, sensor_id)
RETENTIONS 180d
WITH HASH(512);

字段映射很简单：metric -> metric_name、tags -> TAGS、timestamp -> ts、value -> value。脚本必须把秒级和毫秒级时间戳统一成 UTC，不能把原始整数直接写入：

from datetime import datetime, timezone


def to_datetime(timestamp: int) -> datetime:
    seconds = timestamp / 1000 if timestamp >= 10**12 else timestamp
    return datetime.fromtimestamp(seconds, tz=timezone.utc)

基础迁移交给 KDTS，业务转换单独处理

KWDB 3.2.0 已提供基于 DataX 插件的 KDTS 迁移链路：

OpenTSDBReader 2.3.X -> KDTS -> KaiwuDBWriter

它可以完成默认的一指标一表导入：metric 中的 . 转为 _，目标表默认包含 k_timestamp 和 value，表不存在时可以自动创建；使用 beginDateTime 和 endDateTime 时，窗口至少设置为 1 小时。配置只保留结构示意，具体认证和连接字段以安装包中的插件模板为准：

{
  "reader": {
    "name": "opentsdbreader",
    "parameter": {
      "endpoint": "https://opentsdb.example.com",
      "beginDateTime": "2026-10-14 15:34:45",
      "endDateTime": "2026-10-15 12:56:12"
    }
  },
  "writer": {
    "name": "kaiwudbwriter",
    "parameter": {
      "jdbcUrl": "jdbc:kaiwudb://kwdb.example.com:26257/telemetry"
    }
  }
}

如果业务接受默认表结构，KDTS 任务可以直接完成历史数据导入。只有在需要统一长表、保留固定标签，并把 checkpoint、过滤原因和双写补偿纳入同一套流程时，才需要增加转换层。KWDB 的 OpenTSDB JSON 接口用于连通性验证和小批量补偿，成功条件同时看 HTTP 状态和响应体 code = 0。

历史回灌：按小时推进 checkpoint

OpenTSDB 查询结果按序列返回，迁移任务每次读取一个小时。窗口写入成功后才推进 checkpoint，重启时从上一个未确认窗口继续：

import json
from datetime import datetime, timezone
from pathlib import Path

import psycopg
import requests


METRICS = {
    "vehicle.coolant_temp",
    "vehicle.battery_voltage",
    "vehicle.cabin_humidity",
}


def to_datetime(timestamp: int) -> datetime:
    seconds = timestamp / 1000 if timestamp >= 10**12 else timestamp
    return datetime.fromtimestamp(seconds, tz=timezone.utc)


def read_series(base_url: str, token: str, start_s: int, end_s: int):
    payload = {
        "start": start_s,
        "end": end_s,
        "queries": [
            {"aggregator": "none", "metric": metric, "tags": {}}
            for metric in METRICS
        ],
    }
    response = requests.post(
        f"{base_url}/api/query",
        headers={"Authorization": f"Bearer {token}"},
        json=payload,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()


def convert(series):
    rows = []
    for item in series:
        tags = item.get("tags") or {}
        required = ("vehicle_id", "route_id", "sensor_id")
        if item.get("metric") not in METRICS or any(
            not tags.get(key) for key in required
        ):
            continue

        for raw_ts, raw_value in (item.get("dps") or {}).items():
            ts = to_datetime(int(raw_ts))
            rows.append(
                (
                    ts,
                    float(raw_value),
                    item["metric"],
                    str(tags["vehicle_id"]),
                    str(tags["route_id"]),
                    str(tags["sensor_id"]),
                )
            )

    # 同一窗口内以时间、指标和主标签去重。
    return list(
        {
            (row[0], row[2], row[3], row[4], row[5]): row
            for row in rows
        }.values()
    )


def migrate(dsn, source_url, token, checkpoint_path, start_s, end_s):
    checkpoint = Path(checkpoint_path)
    current = start_s
    if checkpoint.exists():
        current = json.loads(checkpoint.read_text(encoding="utf-8"))["next_start_s"]

    with psycopg.connect(dsn, autocommit=True) as conn:
        while current < end_s:
            window_end = min(current + 3600, end_s)
            rows = convert(read_series(source_url, token, current, window_end))

            with conn.cursor() as cur:
                cur.executemany(
                    """
                    INSERT INTO vehicle_metrics
                        (ts, value, metric_name, vehicle_id, route_id, sensor_id)
                    VALUES (%s, %s, %s, %s, %s, %s)
                    """,
                    rows,
                )

            checkpoint.write_text(
                json.dumps({"next_start_s": window_end}),
                encoding="utf-8",
            )
            print(f"window={current}-{window_end}, written={len(rows)}")
            current = window_end

窗口内按 (timestamp, metric, 主标签) 去重，只能解决一次读取返回重复点的问题。跨进程重试仍要保留源数据键，并对边界窗口重新校验。

双写：用 outbox 留住失败事件

历史回灌期间，采集网关继续向 OpenTSDB 写入，同时把原始事件放入本地 outbox。事件只有在两个目标都确认后才标记完成：

CREATE TABLE outbox (
    id INTEGER PRIMARY KEY,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

def flush_one(event, outbox, opentsdb_url, kwdb_dsn):
    write_opentsdb(f"{opentsdb_url}/api/put", event)
    with psycopg.connect(kwdb_dsn, autocommit=True) as conn:
        write_kwdb(conn, event)
    outbox.mark_done(event["id"])

OpenTSDB 已成功、KWDB 失败时，事件保持 pending 并重试；进程在两个写入都成功但状态尚未更新时崩溃，可能产生重复点，所以源数据键和业务侧幂等不能省略。

验收和切换只看四个结果

先比较时间范围和点数，再比较小时聚合和固定抽样点：

SELECT min(ts), max(ts), count(*)
FROM vehicle_metrics
WHERE metric_name = 'vehicle.coolant_temp'
  AND vehicle_id = 'TRUCK-018';

SELECT date_trunc('hour', ts) AS hour_bucket,
       count(*) AS point_count,
       avg(value) AS avg_value
FROM vehicle_metrics
WHERE metric_name = 'vehicle.coolant_temp'
  AND vehicle_id = 'TRUCK-018'
  AND ts >= '2026-01-01 08:07:45+00'
  AND ts <  '2026-01-01 12:50:12+00'
GROUP BY hour_bucket
ORDER BY hour_bucket;

SELECT ts, value
FROM vehicle_metrics
WHERE metric_name = 'vehicle.coolant_temp'
  AND vehicle_id = 'TRUCK-018'
  AND sensor_id = 'coolant-1'
  AND ts = '2026-01-03 12:00:00+00';

切换顺序保持不变：

历史回灌 -> 开启双写 -> 追平增量 -> 影子查询 -> 切读 KWDB
         -> 观察期双写 -> 停止 OpenTSDB 写入

切读前要求最近一小时点数一致、outbox 清空、核心告警查询连续多轮一致。一次演练的验收记录如下：

最后保留三条经验

1. KDTS 已经能完成 OpenTSDB 到 KWDB 的基础导入，先用一小时窗口验证链路，再决定是否需要自定义转换。
2. 默认“一指标一表”不是唯一 Schema。需要统一查询时，先建好长表，再做字段映射，不要让自动建表结果替代设计。
3. checkpoint、outbox 和抽样校验分别解决续跑、失败补偿和结果确认，不能用一个大事务代替它们。

这次切换的完成标准不是“最后一批数据写入成功”，而是同一时间范围内，新旧系统的点数、聚合和抽样结果能够互相解释，并且异常时仍能切回 OpenTSDB。