4、数据存储设计：SQLite数据库设计、创建数据表（测量数据表、日志表、配置表）、使用Python操作SQLite（sqlite3模块）、批量插入与事务处理

说到数据存储，很多初学者第一反应就是「存文件呗」。嗯，存文件确实简单，但你想过没有——当你的万用表每秒采集10个数据点，连续跑一整天，那就是86万条记录。用文本文件存？查个数据得把整个文件读一遍，CPU直接拉满。我个人习惯，这种结构化数据，直接上SQLite。

SQLite是什么？说白了就是一个轻量级的嵌入式数据库。它不需要安装服务器，不需要配置账号密码，一个.db文件搞定一切。Python自带sqlite3模块，开箱即用。我在项目中遇到过很多次，现场环境没有网络、没有数据库服务，全靠SQLite撑起数据记录的重任。

4.1 数据库设计：三张表就够了

设计数据库表，我有个原则：能拆就拆，别把所有东西塞一张表里。你想想看，测量数据、系统日志、设备配置，这三者的读写频率和生命周期完全不同。混在一起，查询性能会越来越差。

我建议分成三张表：

  • 测量数据表（measurements）：存储每次测量的电压、电流、电阻等数值
  • 日志表（logs）：记录系统运行状态、错误信息、操作记录
  • 配置表（config）：保存万用表的量程、采样率、校准参数等

核心设计思路：

  • 测量数据表：高频写入，几乎不修改，按时间查询为主
  • 日志表：中频写入，偶尔查询，用于故障排查
  • 配置表：低频读写，数据量极小，但至关重要

4.2 创建数据表：SQL语句怎么写

直接上代码。这是我实际项目中用过的建表语句，稍微简化了一下：

-- 测量数据表
CREATE TABLE IF NOT EXISTS measurements (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp TEXT NOT NULL,
    channel INTEGER NOT NULL,
    voltage REAL,
    current REAL,
    resistance REAL,
    unit TEXT DEFAULT 'V',
    quality INTEGER DEFAULT 0
);

-- 日志表
CREATE TABLE IF NOT EXISTS logs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp TEXT NOT NULL,
    level TEXT CHECK(level IN ('INFO', 'WARN', 'ERROR')) NOT NULL,
    module TEXT,
    message TEXT NOT NULL
);

-- 配置表
CREATE TABLE IF NOT EXISTS config (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL,
    description TEXT
);

嗯，这里要注意几个细节：

  • timestamp用TEXT类型：SQLite没有专门的日期时间类型，用ISO 8601格式的字符串（比如"2024-01-15 14:30:00"）最方便，排序和过滤都支持
  • quality字段：用来标记数据质量。0表示正常，非零表示异常。我曾经踩过坑，没有这个字段，后来排查数据异常时根本不知道哪些数据是可靠的
  • config表用key-value结构：配置项经常增减，用这种结构最灵活，不用频繁改表结构

避坑指南：

我曾经在生产环境中直接用DROP TABLE重建表，结果把客户的历史数据全清空了。从那以后，我养成了两个习惯：一是建表语句必须加IF NOT EXISTS；二是任何修改表结构的操作，先备份再执行。

4.3 使用Python操作SQLite：三步走

Python操作SQLite其实就三步：连接、执行、关闭。我习惯用上下文管理器，省心又安全。

import sqlite3
from datetime import datetime

# 第一步：连接数据库（文件不存在会自动创建）
conn = sqlite3.connect('multimeter_data.db')

# 第二步：创建游标
cursor = conn.cursor()

# 第三步：执行SQL
cursor.execute('''
    INSERT INTO measurements (timestamp, channel, voltage, unit)
    VALUES (?, ?, ?, ?)
''', (datetime.now().isoformat(), 1, 5.12, 'V'))

# 提交事务
conn.commit()

# 关闭连接
conn.close()

看到那个问号了吗？那是参数化查询的占位符。千万别用字符串拼接的方式拼SQL，容易出SQL注入问题。虽然本地数据库风险小，但养成好习惯总没错。

我的小技巧：

我习惯把数据库操作封装成一个类，这样代码复用性高。比如写一个DatabaseManager，里面包含connect()、insert_measurement()、query_recent()等方法。你想想看，如果每个地方都写一遍连接关闭的代码，后期维护起来多痛苦。

4.4 批量插入与事务处理：性能提升的关键

这是本节的重点。很多新手写数据采集程序，每来一条数据就insert一次。数据量小的时候没问题，但每秒几十上百条数据时，性能会急剧下降。为什么？因为每次insert都涉及磁盘I/O和事务日志。

解决方案很简单：批量插入 + 事务控制。

def batch_insert_measurements(data_list):
    """
    批量插入测量数据
    data_list: 列表，每个元素是 (timestamp, channel, voltage, current, resistance, unit)
    """
    conn = sqlite3.connect('multimeter_data.db')
    cursor = conn.cursor()
    
    try:
        # 开始事务
        cursor.execute('BEGIN TRANSACTION')
        
        # 批量插入
        cursor.executemany('''
            INSERT INTO measurements (timestamp, channel, voltage, current, resistance, unit)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', data_list)
        
        # 提交事务
        conn.commit()
        print(f"成功插入 {len(data_list)} 条数据")
        
    except Exception as e:
        # 出错时回滚
        conn.rollback()
        print(f"插入失败，已回滚：{e}")
        
    finally:
        conn.close()

这段代码里，executemany就是批量插入的核心。它一次接收一个列表，内部循环执行插入操作。配合事务，性能提升非常明显。

我做过一个测试：插入10万条数据，逐条插入耗时约45秒，批量插入（每次1000条）只需要不到2秒。差距就是这么大。

事务处理的黄金法则：

  • 批量操作一定要用事务包裹
  • 事务不要太大，建议每次1000-5000条
  • 务必加上try-except，出错时rollback
  • 长时间运行的程序，定期提交事务，避免事务日志过大

4.5 实际项目中的经验总结

最后分享几个我在项目中踩过的坑：

  • 数据库文件锁问题：SQLite是文件级锁，多线程写入会报错。我一般用一个专门的写入线程，其他线程通过队列把数据传给它
  • 索引很重要：如果经常按时间查询，给timestamp字段加索引。我见过一个项目，没加索引，查询100万条数据花了30秒，加了索引后不到0.1秒
  • 定期清理数据：嵌入式设备存储空间有限，我习惯写一个定时任务，删除30天前的旧数据
  • WAL模式：SQLite的WAL（Write-Ahead Logging）模式能显著提升并发性能。我一般在连接后执行PRAGMA journal_mode=WAL;

嗯，数据存储这块就讲到这里。下一节我们会把这些知识串起来，做一个完整的万用表数据记录系统。到时候你会发现，数据库设计得好，后面的代码写起来特别顺手。

← 上一章 📖 返回目录 下一章 →