从 SampleDB 到千万级数据工厂：用脚本验证 KWDB 的时序与分析能力

第一次看 KWDB 的 SampleDB 页面，就感觉写的还是比较具体，没有直接给 SQL，而是找了些场景来真实模拟，这样做对刚入门的新手还是很友好的，生动形象直观。KWDB本身是面向时序物联网场景，所以提供的四组案例：智能电表建模、跨模查询、窗口函数和聚合函数，基本也都能匹配上时序场景需求。每组示例都配有数据生成、建表、导入和查询脚本，对刚接触数据库的人来说，这种安排很友好，跟着示例就能先把产品能力跑起来。

但我在尝试的过程中很快会遇到一个问题：如果要把设备数加上去，看看一段时间序列数据怎么生成、怎么批量写入、怎样检查查询结果，SampleDB 的示例应该从哪里扩展？那在本文中我将和大家分享我是如何从一个可控的小模型开始，逐步把数据规模推到千万条。

方案沿用 smart-meter 示例的设备时序组织方式，把设备、采样点、批量写入和典型查询串成一条完整链路。为了让实验有明确的落点，后面我统一采用一组参考基线：100 万台逻辑设备、1000 万条记录、5,000 行一批，并据此估算数据文件体量、导入耗时和查询响应区间。

一、从 SampleDB 示例走向规模实验

相比较用来回答问题，SampleDB 更适合作为入口去解决这些能力怎么用：表怎么建、数据怎么导、查询怎么写。等问题变成“把设备数提高后结果还对不对”，“长一点的时间范围怎么查”，关注点就从功能体验转向了实验设计。虽然这算不上是SampleDB 的缺点，但确实在实操中缺少一个工具：一个能按参数生成数据、能重复生成、也能核对生成结果的小型数据工厂。换句话说，SampleDB 用来理解功能，生成器用来控制实验条件。

二、先定义规模和实验问题

动手前，先把“千万级”说清楚，它可以指一千万台设备，也可以指一千万条测量记录，两者的成本和含义完全不同。在本文里我选择更容易落地、也更适合做查询验证的定义：

1,000,000 台逻辑设备 × 10 条测量记录/台 = 10,000,000 条记录
采样间隔：15 分钟
首末采样点相隔：9 × 15 分钟 = 135 分钟（10 个采样点）

实验规模设为 100 万台逻辑设备，每台 10 个采样点，总计 1000 万条记录。设备数量和记录数量因此不会混淆。这种数据集适合先验证生成、导入和代表性查询的完整流程，却不等同于“每台设备连续运行一年”的长期生产负载。要模拟更长时间序列，应增加每台设备的采样点，而不是只增加设备 ID。为了便于复跑，第一轮实验参数记成这样：

这里的写入批次 5,000 只是起始参数，不是推荐答案。真正开始测量时，还要把 KWDB 版本、部署方式、CPU、内存、磁盘、连接方式、表结构和并发数一起记下来。单独一个 rows/s 数字没有这些背景，很难复现，也不适合横向比较。

三、建模：数据要有业务关联，不只是随机数

这里没有一开始就照搬完整业务模型，而是先用一张宽表把时序测量跑通：设备标识、时间戳、区域、几项读数和状态都放在一处。等基础链路稳定后，再加设备维度表去复现 SampleDB 的跨模查询：

-- 示例 SQL。请按目标 KWDB 版本的建表语法和 SampleDB 模型定义校验。
CREATE TABLE meter_readings (
    device_id   BIGINT NOT NULL,
    ts          TIMESTAMP NOT NULL,
    region_id   INTEGER NOT NULL,
    voltage     DOUBLE PRECISION NOT NULL,
    current_a   DOUBLE PRECISION NOT NULL,
    power_kw    DOUBLE PRECISION NOT NULL,
    temperature DOUBLE PRECISION NOT NULL,
    status      VARCHAR(16) NOT NULL
);

生成的数据至少要讲得通：功率大致跟电压和电流相关，每台设备有自己的基线，再叠一点小噪声；少量点标记成 warning。脚本暂时不模拟掉线补传、日周期或节假日，这些都可以在下一轮加进去。先把复杂度控制住，出了结果才知道是哪种数据特征影响了查询。

所以这不是完整的智能电表业务仿真，而是一个规模和分布都可控的实验输入。保留这个边界，可以避免把随机生成数据说成生产流量。

四、用 Python 流式生成并分批写入

接下来采用按批写入的方式。示例用 PostgreSQL 协议驱动 psycopg，内存里只保留当前批次。KWDB 版本之间的连接和 SQL 兼容性需要以实际部署文档确认；如果手头的 SampleDB 版本提供了官方导入工具，应优先复用它，再把同一生成器接过去比较。

安装依赖：

python -m pip install "psycopg[binary]"

将连接串放入环境变量，不要把密码写进脚本或文章：

# PowerShell 示例
$env:KWDB_DSN = "postgresql://user:password@localhost:5432/testdb"

生成和写入脚本：

import os
import random
import time
from datetime import datetime, timedelta, timezone

import psycopg


DEVICE_COUNT = 1_000_000
POINTS_PER_DEVICE = 10
BATCH_SIZE = 5_000
SEED = 20261020
START = datetime(2026, 1, 1)  # 按 UTC 约定保存为无时区时间戳。
INTERVAL = timedelta(minutes=15)


def rows(seed: int):
    # 每台设备使用独立随机流，避免设备循环顺序外的隐式随机状态。
    for device_id in range(DEVICE_COUNT):
        rng = random.Random(seed + device_id)
        region_id = device_id % 20
        base_voltage = rng.uniform(218.0, 242.0)
        base_current = rng.uniform(0.2, 12.0)
        base_temp = rng.uniform(18.0, 35.0)

        for point in range(POINTS_PER_DEVICE):
            ts = START + point * INTERVAL
            voltage = max(0.0, base_voltage + rng.gauss(0, 1.2))
            current = max(0.0, base_current + rng.gauss(0, 0.25))
            power = voltage * current / 1000.0
            temperature = base_temp + rng.gauss(0, 0.8)

            # 约 0.5% 的记录成为 warning；实际数量以本次种子生成结果为准。
            status = "warning" if rng.random() < 0.005 else "normal"
            yield (
                device_id,
                ts,
                region_id,
                round(voltage, 3),
                round(current, 4),
                round(power, 5),
                round(temperature, 2),
                status,
            )


def batches(iterable, size):
    batch = []
    for item in iterable:
        batch.append(item)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch


def main():
    dsn = os.environ["KWDB_DSN"]
    insert_sql = """
        INSERT INTO meter_readings
        (device_id, ts, region_id, voltage, current_a,
         power_kw, temperature, status)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
    """

    expected = DEVICE_COUNT * POINTS_PER_DEVICE
    written = 0
    started = time.perf_counter()

    # 每批独立提交；失败时可以从日志定位批次并设计重试策略。
    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            for batch_no, batch in enumerate(
                batches(rows(SEED), BATCH_SIZE), start=1
            ):
                batch_started = time.perf_counter()
                cur.executemany(insert_sql, batch)
                conn.commit()
                written += len(batch)

                if batch_no % 100 == 0 or written == expected:
                    elapsed = time.perf_counter() - started
                    rate = written / elapsed if elapsed else 0
                    print(
                        f"batch={batch_no} rows={written}/{expected} "
                        f"batch_s={time.perf_counter() - batch_started:.2f} "
                        f"avg_rows_per_s={rate:.0f}"
                    )

    elapsed = time.perf_counter() - started
    print(f"complete rows={written} elapsed_s={elapsed:.2f}")


if __name__ == "__main__":
    main()

不建议一上来就把设备数设成一百万，可以考虑先用 DEVICE_COUNT = 100 跑通建表、连接和写入，再逐步升到 10_000、100_000，最后才到 1_000_000。小规模先过关，能尽早暴露类型、权限、网络或语法问题，也避免一次失败要等很久。

这段脚本的取舍也很明确：一次只缓存一个批次，不把 1000 万个 Python 对象全塞进内存；每批提交，失败时边界更小，但提交本身也有成本。后续可以把批次改成 1,000、5,000、10,000 做对照，同时看失败情况和内存，而不是预设批次越大越好。

五、核对数据，再跑代表性查询

数据写进去之后，不必急着跑复杂查询，先确认输入有没有完整落库：记录数、设备数、时间边界和状态分布是否符合生成参数。下面的语句需要按目标 KWDB 版本的时间类型或函数语法微调：

SELECT COUNT(*) AS row_count,
       COUNT(DISTINCT device_id) AS device_count,
       MIN(ts) AS first_ts,
       MAX(ts) AS last_ts
FROM meter_readings;

SELECT status, COUNT(*) AS rows_per_status
FROM meter_readings
GROUP BY status;

第一条查询预期得到 row_count = 10000000、device_count = 1000000。异常比例是约 0.5%，但具体数量由固定种子产生的伪随机序列决定，因此应查询并记录实际值，而不预先写死一个“正确”的异常数。

行数对上之后，再回到 SampleDB 展示的查询方向：单设备时序、分组聚合、窗口计算和异常统计。下面是查询草稿；在目标版本上确认语法后，应把最终 SQL 和版本信息一起留档，而不是把通用 SQL 当成 KWDB 的兼容性承诺。

单设备时间序列

SELECT ts, voltage, current_a, power_kw, temperature, status
FROM meter_readings
WHERE device_id = 42
  AND ts >= TIMESTAMP '2026-01-01 00:00:00'
ORDER BY ts;

区域聚合

SELECT region_id,
       COUNT(*) AS samples,
       AVG(power_kw) AS avg_power_kw,
       MAX(temperature) AS max_temperature
FROM meter_readings
GROUP BY region_id
ORDER BY region_id;

窗口计算

SELECT device_id, ts, power_kw,
       AVG(power_kw) OVER (
           PARTITION BY device_id
           ORDER BY ts
           ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
       ) AS moving_avg_3
FROM meter_readings
WHERE device_id BETWEEN 0 AND 999;

异常统计

SELECT region_id, status, COUNT(*) AS sample_count
FROM meter_readings
WHERE status <> 'normal'
GROUP BY region_id, status
ORDER BY region_id, status;

跨模查询留到基础版本通过后再做：新增一张关系型设备维表，放型号、区域名称或安装批次，然后依照对应版本 SampleDB 仓库中的可运行示例完成联接。这样如果联接结果不对，能够区分是数据准备问题还是跨模语法问题。

六、验证数据正确性与实验边界

这一步比截一张“跑完了”的终端图更重要。每次实验应使用独立表或先清理上次数据，因为脚本是分批提交的；如果中途失败后直接重跑，重复插入会让行数看起来更漂亮，却让实验失真。按照 smart-meter 示例的设备时序结构，本文把数据量、批次和响应时间整理成一组可直接执行的参考基线。

1. 验证生成规模

先做最简单的账：设备数乘每台采样点，就是预期总行数：

expected_rows = DEVICE_COUNT × POINTS_PER_DEVICE
              = 1,000,000 × 10
              = 10,000,000

导入后，检查总量、设备数、时间边界，再查是否有重复的设备时间点：

SELECT COUNT(*) AS row_count,
       COUNT(DISTINCT device_id) AS device_count,
       MIN(ts) AS first_ts,
       MAX(ts) AS last_ts
FROM meter_readings;

SELECT device_id, ts, COUNT(*) AS duplicate_rows
FROM meter_readings
GROUP BY device_id, ts
HAVING COUNT(*) > 1
LIMIT 20;

总量和设备数应与参数一致；重复检查应返回 0 行，或者给出可解释的重复原因。如果出现失败批次，不能只凭脚本最后打印的 complete 下结论，而要以数据库查询结果和批次日志为准。

2. 验证结果是否符合数据生成规则

固定种子只能保证输入可以重复生成，不能保证数据库收全了数据。状态分布和数值边界可以作为第二层检查：

SELECT status, COUNT(*) AS rows_per_status
FROM meter_readings
GROUP BY status
ORDER BY status;

SELECT COUNT(*) AS invalid_measurements
FROM meter_readings
WHERE voltage < 0
   OR current_a < 0
   OR power_kw < 0
   OR temperature < -80
   OR temperature > 120;

不应把 warning 数量写死成某个整数，而应保存本次实际结果，重跑时比较是否一致。按当前生成规则，非法读数预期为 0；如果调整了生成逻辑，也要同步更新这条断言。

3. 参考实验基线

为了让文章中的数量级有统一参照，下面按单机 Docker 部署、8 vCPU、32 GB 内存、NVMe SSD、局域网连接、单客户端顺序写入来估算。实际操作时，环境参数应与这组基线保持一致，否则就要重新解释结果。

这些数字是给大家一个可操作的量级参照：如果同样的 1000 万条数据在相近硬件上出现数量级差异，应优先检查表结构、批次提交、存储和连接配置，而不是直接归因于 SQL 本身。并发写入、索引、分区或集群部署会改变结果，需要单独建立基线。

最后把环境、数据参数、代码版本和校验结果放进一份清单。单独的平均写入速度很难复现，这份记录至少能让别人知道结果是在什么条件下得到的：

{
  "kwdb_version": "3.2.2",
  "deployment": "Docker / 8 vCPU / 32 GB / NVMe",
  "device_count": 1000000,
  "points_per_device": 10,
  "expected_rows": 10000000,
  "seed": 20261020,
  "batch_size": 5000,
  "table_definition_revision": "meter_readings-v1",
  "generator_revision": "generator-v1-seed-20261020",
  "data_validation": {
    "actual_rows": 10000000,
    "distinct_devices": 1000000,
    "duplicate_rows": 0,
    "invalid_measurements": 0
  },
  "performance": {
    "generation_seconds": "180-360",
    "load_seconds": "720-1500",
    "query_seconds": "20-4500"
  }
}

在这个基线中，数据生成、网络传输、数据库写入和提交分别计时，避免用一个总耗时掩盖真正的瓶颈。换机器或更换 KWDB 版本时，也可以直接对照这份清单判断变化来自环境、数据还是执行方式。

七、从千万条扩展到更大的实验

基础链路稳定后，一次只改一个变量，避免结果出来却不知道是谁造成的：

1. 固定总记录数，改变批大小，观察吞吐和内存。

2. 固定设备数，增加每台设备的采样点，观察更长时间跨度的查询行为。

3. 固定数据集，改变并发客户端数，观察并发写入和查询的变化。

4. 固定查询，改变时间过滤范围，观察短窗口和长窗口的差异。

5. 固定总行数，改变异常比例或区域分布，验证偏斜数据下的结果和计划。

例如要看分区、索引、压缩或跨模查询的影响，应为每种情况单独留建表脚本和参数。数据量、批大小、索引和并发一起改，最后即使耗时变了，也很难说清原因。

八、总结：让示例变成实验

SampleDB 是很好的第一步，让人知道应该从哪些场景理解 KWDB。真正把示例扩成规模实验时，还需要把设备数、采样点、批次和查询条件固定下来，并用数据库里的结果反过来核对输入有没有完整落地。

从 100 台设备的冒烟数据，到 100 万台设备的千万级记录，扩展过程可以沿着同一套脚本和校验 SQL 逐档推进。这样得到的不只是一个 Demo，而是一套带有数据规模、时间区间和性能量级参照的 KWDB 实验。