基于 KaiwuDB 社区版的车联网时序数据存储实战：从踩坑到全链路部署部署测试

前言

最近团队在做一个智能网联汽车（IoV）的项目，随着接入的车辆越来越多，后端的存储压力肉眼可见地变大。现在的车跟以前不一样，车上的 CAN 总线、GPS、各类传感器每秒钟都在往云端发数据。

起初我们用的是传统的关系型数据库，但很快就撑不住了。于是我花了一些时间去调研专门针对物联网和时序数据的数据库，最终拿 KWDB（KaiwuDB 社区版）做了一次从零开始的完整测试。这篇文章就算是我这次技术预研的一份笔记，从底层环境搭建、数据库部署、表结构设计，一直到 Python 写脚本压测和出图，希望能给遇到类似存储瓶颈的兄弟们一点参考。

---

第一章：我们在车联网数据存储上遇到的坑

在直接上手干活之前，先聊聊我们实际业务中遇到的几个痛点，这也是为什么我要换数据库的原因。

1.1 并发写入量太大

车联网最明显的特点就是高频写入。假设平台有 10 万辆车，每辆车每秒传 10 个数据点（位置、速度、胎压等），那数据库每秒就要扛 100 万次写入。这种压力是持续的，不像电商那样一阵一阵的。传统的 B+ 树索引数据库面对这种持续的随机/顺序写入，频繁的锁竞争和页分裂会让性能直接拉胯，数据积压严重。

1.2 磁盘空间不够用

一辆车一天产生几十上百兆数据，车一多，每天就是 TB 甚至 PB 级的数据量。如果不做极致的压缩，光买硬盘的钱公司都受不了。而且车联网数据“喜新厌旧”，最近几天的数据经常查，半年前的数据基本吃灰。我们需要数据库本身就有很强的压缩率，并且能方便地处理冷热数据。

1.3 复杂的组合查询

业务端的查询需求往往很变态，比如：“查一下朝阳区过去一小时内，平均车速低于 20km/h 的所有车”。这就要求数据库得同时搞定时间范围、空间坐标和普通字段的聚合计算。以前我们是用 Redis + MySQL + ES 组合来搞，架构太臃肿，维护起来心力交瘁。

基于这几个原因，我挑了带有列式压缩、时序聚合优化且底层用 LSM-Tree 存储引擎的 KWDB 来做这次测试。

---

第二章：系统环境准备和依赖安装

我的测试机是一台装了 Ubuntu 22.04 LTS 的普通服务器。由于 KWDB 底层依赖一些特定的系统 C++ 库，所以在装数据库之前，得先把环境理顺。

2.1 安装核心依赖库

直接用 apt 把需要的库全装上：

sudo apt update 
sudo apt install -y openssl libprotobuf-dev libgeos-dev liblz4-dev xz-utils libgcc-9-dev libgflags-dev libkrb5-dev

这里简单提几个关键的库：

• libprotobuf-dev：车联网高频写入对网络带宽消耗很大，KWDB 底层用 Protobuf 做序列化，比传 JSON 体积小、解析快。
• libgeos-dev：处理经纬度、电子围栏这些空间计算必须的几何算法库。
• liblz4-dev：LZ4 压缩算法。时序数据（比如连续几秒的车速）相似度很高，用这个库在内存和磁盘里做实时压缩，能省下不少空间。

2.2 确认依赖版本

装完后最好检查一下版本，特别是 Protobuf，官方要求是 >= 3.6.1，不然跑不起来。

dpkg -l | grep libprotobuf

终端打印出来的版本是 libprotobuf-dev 3.12.4，没问题，可以继续。

2.3 防火墙端口开放

测试环境为了方便，我直接在 UFW 里开了两个必须的端口：

• 8080：KWDB 自带的 Web 监控控制台端口，浏览器看监控用的。
• 26257：数据库对外提供服务的监听端口，应用连数据库就填这个。

---

第三章：数据库部署与单机配置

环境没问题后，就可以开始装数据库了。

3.1 下载和上传包

去官网下载对应 CPU 架构（我是 x86_64）的 DEB 安装包，然后传到服务器的目录下解压。

3.2 改配置（单机性能压榨）

解压后，目录里有个 deploy.cfg 文件。KWDB 默认配置是为分布式集群准备的，但我这次是为了测单机的极限写入性能。如果开着集群模式，底层会有 Raft 协议的额外消耗。

所以我把配置文件里 [cluster] 和 [additional] 相关的段落直接注释掉了，强制让它以单节点 (Single Node) 模式跑。

3.3 启动服务

带上 --single 参数跑安装和启动脚本：

./deploy.sh install --single

完事儿后看下状态：

./deploy.sh status

看到 Status 是 running，进程 PID 也出来了，说明数据库已经正常跑起来了。

---

第四章：建表与 Python 模拟数据写入

数据库搭好了，接下来搞点代码。我习惯用 Python 做测试，因为 KWDB 兼容 PostgreSQL 协议，所以直接用现成的 PG 驱动就行，不需要学新的语法。

4.1 配置文件

在本地写个 config.py，把数据库连接参数抽出来，这样后面改 IP 或者密码比较方便。

4.2 安装 Python 库

测试脚本需要用到这几个库：

• psycopg2-binary：连数据库的驱动。
• pandas & matplotlib：等下用来查数据和画图。

4.3 表结构设计

新建一个 01_create_tables.py 脚本来建表。结合车联网的特点，我建了两张表：一张存车辆轨迹，一张存传感器数据。

这里有个关键的细节：主键的顺序。我把主键设为 PRIMARY KEY (vehicle_id, ts)。把车牌号放前面，时间戳放后面，这样同一辆车的数据在物理磁盘上是挨着存的，等下查单车历史轨迹的时候速度会非常快。

用客户端连进去看一眼，表建成功了。

"""
创建车联网数据表
"""
import psycopg2
from config import KWDB_CONFIG

def create_tables():
    conn = psycopg2.connect(**KWDB_CONFIG)
    cursor = conn.cursor()

    # 创建车辆位置表
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS vehicle_location (
        ts TIMESTAMPTZ NOT NULL,
        vehicle_id VARCHAR(50) NOT NULL,
        longitude DOUBLE PRECISION,
        latitude DOUBLE PRECISION,
        speed DOUBLE PRECISION,
        direction INT,
        PRIMARY KEY (vehicle_id, ts)
    );
    """)

    # 创建传感器数据表
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS vehicle_sensor (
        ts TIMESTAMPTZ NOT NULL,
        vehicle_id VARCHAR(50) NOT NULL,
        engine_temp DOUBLE PRECISION,
        oil_pressure DOUBLE PRECISION,
        battery_voltage DOUBLE PRECISION,
        fuel_level DOUBLE PRECISION,
        PRIMARY KEY (vehicle_id, ts)
    );
    """)

    conn.commit()
    cursor.close()
    conn.close()
    print("✓ 表创建成功")

if __name__ == "__main__":
    create_tables()

4.4 模拟数据批量写入

为了测试，我写了个 02_insert_data.py 脚本造点假数据。代码模拟了车辆在北京附近的移动，经纬度和传感器数值加了随机波动。

重点是写入方式：绝对不能写个 for 循环一条条 INSERT，那样光网络通信的时间就把性能耗光了。我用了 psycopg2 的 executemany，把数据攒成几千条一个批次发给数据库。

运行了一下，控制台打印出结果了。

再去数据库里 GROUP BY 查一下，数据已经落盘了，分布也很均匀。

"""
生成并批量插入车联网模拟数据
"""
import psycopg2
from datetime import datetime, timedelta
import random
from config import KWDB_CONFIG

def generate_vehicle_data(vehicle_id, start_time, count=1000):
    """生成模拟车辆数据"""
    location_data = []
    sensor_data = []

    # 模拟起点：北京市中心
    base_lng = 116.4074
    base_lat = 39.9042

    for i in range(count):
        ts = (start_time + timedelta(seconds=i*10)).strftime('%Y-%m-%d %H:%M:%S')

        # 位置数据：模拟车辆移动
        location_data.append((
            ts,
            vehicle_id,
            base_lng + random.uniform(-0.1, 0.1),
            base_lat + random.uniform(-0.1, 0.1),
            random.uniform(0, 120),  # 速度
            random.randint(0, 360)   # 方向
        ))

        # 传感器数据
        sensor_data.append((
            ts,
            vehicle_id,
            random.uniform(80, 110),   # 发动机温度
            random.uniform(2.5, 4.5),  # 机油压力
            random.uniform(12, 14.5),  # 电池电压
            random.uniform(10, 100)    # 油量百分比
        ))

    return location_data, sensor_data

def batch_insert(vehicle_count=10, records_per_vehicle=1000):
    """批量插入数据"""
    conn = psycopg2.connect(**KWDB_CONFIG)
    cursor = conn.cursor()

    start_time = datetime.now() - timedelta(hours=3)

    print(f"开始生成 {vehicle_count} 辆车的数据，每辆车 {records_per_vehicle} 条记录...")

    total_records = 0
    insert_start = datetime.now()

    for i in range(vehicle_count):
        vehicle_id = f"CAR{i+1:04d}"
        location_data, sensor_data = generate_vehicle_data(
            vehicle_id, start_time, records_per_vehicle
        )

        # 批量插入位置数据
        cursor.executemany("""
            INSERT INTO vehicle_location (ts, vehicle_id, longitude, latitude, speed, direction)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, location_data)

        # 批量插入传感器数据
        cursor.executemany("""
            INSERT INTO vehicle_sensor (ts, vehicle_id, engine_temp, oil_pressure, battery_voltage, fuel_level)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, sensor_data)

        total_records += len(location_data) + len(sensor_data)

        # 每5辆车提交一次
        if (i + 1) % 5 == 0:
            conn.commit()
            print(f"已处理 {i+1}/{vehicle_count} 辆车")

    conn.commit()
    insert_end = datetime.now()
    duration = (insert_end - insert_start).total_seconds()

    print(f"\n✓ 数据插入完成")
    print(f"  总记录数: {total_records:,}")
    print(f"  耗时: {duration:.2f} 秒")
    print(f"  写入速度: {total_records/duration:,.0f} 条/秒")

    cursor.close()
    conn.close()

if __name__ == "__main__":
    batch_insert(vehicle_count=10, records_per_vehicle=1000)

---

第五章：常用查询场景测试与出图

数据存进去了，关键还得看查得快不快。我针对我们平时用的最多的几个业务场景写了个 03_query_tests.py 测试脚本。

5.1 四个查询场景测试

跑出来的耗时结果都很令人满意：

1. 单车轨迹（查特定 ID 和时间）：因为主键优化的原因，不用扫全表，瞬间返回。
2. 车队速度均值（AVG 等聚合查询）：列式存储的优势出来了，它只读 speed 这一列的数据去算，避开了没用的列，速度很快。
3. 异常报警（多个条件 OR 过滤）：像温度高或胎压低的过滤，直接推到存储层完成了。
4. 降采样（5分钟聚合）：用 Postgres 的 date_trunc 搞定，算是时序数据的常规操作。

"""
车联网数据查询场景测试
"""
import psycopg2
from datetime import datetime, timedelta
import pandas as pd
from config import KWDB_CONFIG

def test_query_1_vehicle_trajectory():
    """场景1：查询单车轨迹"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT ts, longitude, latitude, speed
    FROM vehicle_location
    WHERE vehicle_id = 'CAR0001'
    AND ts >= NOW() - INTERVAL '1 hour'
    ORDER BY ts
    LIMIT 100;
    """

    start = datetime.now()
    df = pd.read_sql(query, conn)
    duration = (datetime.now() - start).total_seconds()

    print("=" * 60)
    print("场景1：单车轨迹查询")
    print("=" * 60)
    print(f"查询条件: 车辆 CAR0001，最近1小时")
    print(f"返回记录: {len(df)} 条")
    print(f"查询耗时: {duration*1000:.2f} ms")
    print(f"\n前5条数据:")
    print(df.head())

    conn.close()
    return df

def test_query_2_speed_statistics():
    """场景2：车速统计分析"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT
        vehicle_id,
        AVG(speed) as avg_speed,
        MAX(speed) as max_speed,
        MIN(speed) as min_speed
    FROM vehicle_location
    WHERE ts >= NOW() - INTERVAL '1 hour'
    GROUP BY vehicle_id
    ORDER BY avg_speed DESC;
    """

    start = datetime.now()
    df = pd.read_sql(query, conn)
    duration = (datetime.now() - start).total_seconds()

    print("\n" + "=" * 60)
    print("场景2：车速统计分析")
    print("=" * 60)
    print(f"统计范围: 所有车辆，最近1小时")
    print(f"返回记录: {len(df)} 条")
    print(f"查询耗时: {duration*1000:.2f} ms")
    print(f"\n统计结果:")
    print(df)

    conn.close()
    return df

def test_query_3_sensor_anomaly():
    """场景3：传感器异常检测"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT
        ts, vehicle_id, engine_temp, oil_pressure, battery_voltage
    FROM vehicle_sensor
    WHERE ts >= NOW() - INTERVAL '30 minutes'
    AND (engine_temp > 105 OR oil_pressure < 3.0 OR battery_voltage < 12.5)
    ORDER BY ts DESC
    LIMIT 20;
    """

    start = datetime.now()
    df = pd.read_sql(query, conn)
    duration = (datetime.now() - start).total_seconds()

    print("\n" + "=" * 60)
    print("场景3：传感器异常检测")
    print("=" * 60)
    print(f"检测条件: 发动机温度>105℃ 或 机油压力<3.0 或 电池电压<12.5V")
    print(f"异常记录: {len(df)} 条")
    print(f"查询耗时: {duration*1000:.2f} ms")
    if len(df) > 0:
        print(f"\n异常数据样例:")
        print(df.head(10))

    conn.close()
    return df

def test_query_4_time_window():
    """场景4：时间窗口聚合"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT
        date_trunc('minute', ts) as time_window,
        vehicle_id,
        AVG(speed) as avg_speed,
        COUNT(*) as record_count
    FROM vehicle_location
    WHERE vehicle_id = 'CAR0001'
    AND ts >= NOW() - INTERVAL '1 hour'
    GROUP BY time_window, vehicle_id
    ORDER BY time_window;
    """

    start = datetime.now()
    df = pd.read_sql(query, conn)
    duration = (datetime.now() - start).total_seconds()

    print("\n" + "=" * 60)
    print("场景4：5分钟时间窗口聚合")
    print("=" * 60)
    print(f"聚合对象: 车辆 CAR0001")
    print(f"返回记录: {len(df)} 条")
    print(f"查询耗时: {duration*1000:.2f} ms")
    print(f"\n聚合结果:")
    print(df)

    conn.close()
    return df

if __name__ == "__main__":
    test_query_1_vehicle_trajectory()
    test_query_2_speed_statistics()
    test_query_3_sensor_anomaly()
    test_query_4_time_window()
    print("\n✓ 所有查询测试完成")

5.2 跑点图表看看

光看命令行数据不直观，我顺手写了个 Python 画图脚本，把前面查出来的数据喂给 matplotlib。

1. 传感器监控看板
把发动机温度、机油压力和电池电压画出来了，加了条红色的警戒线。以后在前端页面照着这个做，谁的数据快超标了一目了然。

2. 车速折线图
很直观地看到这辆车刚才的加减速情况。

3. 轨迹+速度散点图
把经纬度当坐标轴，点的颜色深浅代表速度。这就类似我们在打车软件后台看到的热力图了。

展示绘图脚本的源码：

"""
数据可视化展示
"""
import psycopg2
import pandas as pd
import matplotlib.pyplot as plt
from config import KWDB_CONFIG

plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def visualize_trajectory():
    """可视化车辆轨迹"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT longitude, latitude, speed
    FROM vehicle_location
    WHERE vehicle_id = 'CAR0001'
    AND ts >= NOW() - INTERVAL '1 hour'
    ORDER BY ts
    LIMIT 500;
    """

    df = pd.read_sql(query, conn)
    conn.close()

    plt.figure(figsize=(10, 6))
    scatter = plt.scatter(df['longitude'], df['latitude'],
                         c=df['speed'], cmap='RdYlGn_r', s=20, alpha=0.6)
    plt.colorbar(scatter, label='速度 (km/h)')
    plt.xlabel('经度')
    plt.ylabel('纬度')
    plt.title('车辆轨迹热力图 (CAR0001)')
    plt.grid(True, alpha=0.3)
    plt.savefig('trajectory.png', dpi=150, bbox_inches='tight')
    print("✓ 轨迹图已保存: trajectory.png")

def visualize_speed_trend():
    """可视化速度趋势"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT ts, speed
    FROM vehicle_location
    WHERE vehicle_id = 'CAR0001'
    AND ts >= NOW() - INTERVAL '1 hour'
    ORDER BY ts;
    """

    df = pd.read_sql(query, conn)
    conn.close()

    plt.figure(figsize=(12, 5))
    plt.plot(df['ts'], df['speed'], linewidth=1, alpha=0.7)
    plt.xlabel('时间')
    plt.ylabel('速度 (km/h)')
    plt.title('车辆速度变化趋势')
    plt.grid(True, alpha=0.3)
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig('speed_trend.png', dpi=150, bbox_inches='tight')
    print("✓ 速度趋势图已保存: speed_trend.png")

def visualize_sensor_data():
    """可视化传感器数据"""
    conn = psycopg2.connect(**KWDB_CONFIG)

    query = """
    SELECT ts, engine_temp, oil_pressure, battery_voltage
    FROM vehicle_sensor
    WHERE vehicle_id = 'CAR0001'
    AND ts >= NOW() - INTERVAL '1 hour'
    ORDER BY ts;
    """

    df = pd.read_sql(query, conn)
    conn.close()

    fig, axes = plt.subplots(3, 1, figsize=(12, 10))

    axes[0].plot(df['ts'], df['engine_temp'], color='red', linewidth=1)
    axes[0].set_ylabel('发动机温度 (℃)')
    axes[0].grid(True, alpha=0.3)
    axes[0].axhline(y=105, color='r', linestyle='--', alpha=0.5, label='警戒线')
    axes[0].legend()

    axes[1].plot(df['ts'], df['oil_pressure'], color='blue', linewidth=1)
    axes[1].set_ylabel('机油压力 (bar)')
    axes[1].grid(True, alpha=0.3)
    axes[1].axhline(y=3.0, color='r', linestyle='--', alpha=0.5, label='警戒线')
    axes[1].legend()

    axes[2].plot(df['ts'], df['battery_voltage'], color='green', linewidth=1)
    axes[2].set_ylabel('电池电压 (V)')
    axes[2].set_xlabel('时间')
    axes[2].grid(True, alpha=0.3)
    axes[2].axhline(y=12.5, color='r', linestyle='--', alpha=0.5, label='警戒线')
    axes[2].legend()

    plt.suptitle('传感器数据监控')
    plt.tight_layout()
    plt.savefig('sensor_monitor.png', dpi=150, bbox_inches='tight')
    print("✓ 传感器监控图已保存: sensor_monitor.png")

if __name__ == "__main__":
    visualize_trajectory()
    visualize_speed_trend()
    visualize_sensor_data()
    print("\n✓ 所有可视化图表生成完成")

---

第六章：最后来一波基准压测

功能跑通了，最后肯定要测测极限性能。我写了一个简单的压测脚本，分别测试了不同批次大小（Batch Size）下的写入 TPS，以及刚才那几个查询场景的真实延迟。

"""
性能测试：对比不同数据量下的写入和查询性能
"""
import psycopg2
from datetime import datetime, timedelta
import random
from config import KWDB_CONFIG

def performance_test_write(batch_sizes=[100, 500, 1000, 5000]):
    """测试不同批次大小的写入性能"""
    conn = psycopg2.connect(**KWDB_CONFIG)
    cursor = conn.cursor()

    print("=" * 60)
    print("写入性能测试")
    print("=" * 60)

    # 清理之前的测试数据
    cursor.execute("DELETE FROM vehicle_location WHERE vehicle_id = 'TEST_CAR'")
    conn.commit()

    results = []

    for batch_size in batch_sizes:
        # 生成测试数据，使用微秒确保唯一性
        data = []
        base_time = datetime.now()
        for i in range(batch_size):
            data.append((
                (base_time + timedelta(seconds=i, microseconds=i*1000)).strftime('%Y-%m-%d %H:%M:%S.%f'),
                'TEST_CAR',
                116.4 + random.uniform(-0.1, 0.1),
                39.9 + random.uniform(-0.1, 0.1),
                random.uniform(0, 120),
                random.randint(0, 360)
            ))

        # 执行写入
        start = datetime.now()
        cursor.executemany("""
            INSERT INTO vehicle_location (ts, vehicle_id, longitude, latitude, speed, direction)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, data)
        conn.commit()
        duration = (datetime.now() - start).total_seconds()

        throughput = batch_size / duration
        results.append({
            'batch_size': batch_size,
            'duration': duration,
            'throughput': throughput
        })

        print(f"批次大小: {batch_size:5d} | 耗时: {duration:.3f}s | 吞吐量: {throughput:,.0f} 条/秒")

    # 清理测试数据
    cursor.execute("DELETE FROM vehicle_location WHERE vehicle_id = 'TEST_CAR'")
    conn.commit()

    cursor.close()
    conn.close()

    return results

def performance_test_query():
    """测试不同查询场景的性能"""
    conn = psycopg2.connect(**KWDB_CONFIG)
    cursor = conn.cursor()

    print("\n" + "=" * 60)
    print("查询性能测试")
    print("=" * 60)

    queries = [
        ("点查询", "SELECT * FROM vehicle_location WHERE vehicle_id = 'CAR0001' AND ts >= NOW() - INTERVAL '10 minutes' LIMIT 100"),
        ("聚合查询", "SELECT vehicle_id, AVG(speed) FROM vehicle_location WHERE ts >= NOW() - INTERVAL '1 hour' GROUP BY vehicle_id"),
        ("范围扫描", "SELECT * FROM vehicle_location WHERE ts >= NOW() - INTERVAL '2 hours' LIMIT 1000"),
        ("多表关联", """
            SELECT l.vehicle_id, l.speed, s.engine_temp
            FROM vehicle_location l
            JOIN vehicle_sensor s ON l.vehicle_id = s.vehicle_id AND l.ts = s.ts
            WHERE l.ts >= NOW() - INTERVAL '30 minutes'
            LIMIT 100
        """)
    ]

    for name, query in queries:
        start = datetime.now()
        cursor.execute(query)
        results = cursor.fetchall()
        duration = (datetime.now() - start).total_seconds()

        print(f"{name:10s} | 返回: {len(results):5d} 条 | 耗时: {duration*1000:6.2f} ms")

    cursor.close()
    conn.close()

if __name__ == "__main__":
    performance_test_write()
    performance_test_query()
    print("\n✓ 性能测试完成")

压测结果简单分析

这是跑出来的真实结果：

1. 写入部分：如果每次只发 100 条（小批次），吞吐量在 25,000 条/秒左右，这里很大一部分时间耗在网络和握手上。当把批次拉大到 5000 条一次时，吞吐量突破了 27,000 条/秒。单机能扛住两万七的并发写入，应付我们目前几十万辆在线车的规模（每 10 秒报一次点）绰绰有余了。
2. 查询部分：
   • 点查询只要 8.45 毫秒，基本上秒回。
   • 聚合查询（算 AVG）也只花了 12.35 毫秒，这点让我比较惊喜。
   • 连平时最容易卡脖子的两表 Join，耗时也控制在了 18 毫秒出头。

第七章：总结

折腾了这几天，总结一下这次用 KWDB 社区版做车联网数据后端的感受：

1. 写入确实能打：省去了频繁改配置和调优的麻烦，单机拉起直接测就有 2.7 万的 TPS。
2. 不用学新东西：完全兼容 PostgreSQL 协议是个巨大的加分项。我们原本的 Python 代码基本不用怎么改，Pandas 这些包也都能无缝对接。
3. 统计查询快：受惠于列存架构，按列去做平均值、最大值这类统计查询，比以前用 MySQL 快太多了。

对于正在被海量 IoT 设备日志或者车联网数据折磨的开发兄弟，如果现有的技术栈扛不住了，可以试试引入专门的时序数据库。这次单机版的测试验证了可行性，下一步我会试着在内网搭一个三节点的分布式集群，测一下高可用和容灾能力。