4、数据存储：将清洗后的数据存入本地SQLite数据库，设计表结构

数据清洗完了，下一步就是存起来。

你可能觉得，存数据嘛，不就是 insert 一下？其实没那么简单。表结构设计得好不好，直接影响你后面查数据、做报表、甚至上云同步的效率。我刚开始做物联网项目时，就吃过这个亏——表结构拍脑袋就定了，结果后面改得想哭。

4.1 为什么选 SQLite？

说实话，本地存储的选择不少。CSV、JSON、Excel 都能存。但为什么我坚持用 SQLite？

  • 零配置：不需要装数据库服务，一个文件搞定
  • 标准 SQL：你会的 SQL 语法，它基本都支持
  • Python 内置支持：import sqlite3 就能用，不用装第三方库
  • 轻量级：整个库才几百 KB，嵌入式设备也能跑

嗯，这里要注意一点：SQLite 不适合高并发写入场景。但咱们这个课程是单机采集、单机存储，完全够用。我在一个树莓派项目上跑过 SQLite，连续采集了三个月，稳得很。

4.2 表结构设计思路

设计表结构，说白了就是回答三个问题：

  1. 要存什么？
  2. 怎么查最快？
  3. 以后怎么扩展？

咱们的空气质量数据，核心字段包括：

  • 采集时间（精确到秒）
  • PM2.5 浓度
  • PM10 浓度
  • 温度
  • 湿度
  • 设备编号（如果有多个设备）

你想想看，如果以后要按天、按小时做统计，时间字段就特别重要。所以我习惯把时间设为主键的一部分，或者至少建个索引。

核心原则：宁可多花一分钟设计表结构，不要花一小时改数据。

4.3 建表 SQL 语句

直接上代码。这是我个人比较喜欢的一种设计：

CREATE TABLE IF NOT EXISTS air_quality (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    device_id TEXT NOT NULL DEFAULT 'device_001',
    record_time TEXT NOT NULL,
    pm25 REAL,
    pm10 REAL,
    temperature REAL,
    humidity REAL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(device_id, record_time)
);

解释一下几个关键点：

  • id：自增主键，方便按行操作
  • device_id：预留多设备场景，默认单设备
  • record_time：采集时间，用 TEXT 类型存 ISO 格式字符串
  • UNIQUE 约束：同一个设备同一时间只能有一条记录，防止重复插入

小技巧：record_time 用 TEXT 而不是 DATETIME 类型，是因为 SQLite 没有真正的日期时间类型。用 ISO 格式字符串（如 "2024-01-15 08:30:00"）排序、比较都方便。

4.4 Python 实现：连接与建表

代码写起来其实很简单。我习惯把数据库操作封装成一个类，方便复用：

import sqlite3
import os

class AirQualityDB:
    def __init__(self, db_path="air_quality.db"):
        self.db_path = db_path
        self.conn = None
        self.cursor = None

    def connect(self):
        """连接数据库，如果文件不存在会自动创建"""
        self.conn = sqlite3.connect(self.db_path)
        self.cursor = self.conn.cursor()
        print(f"✅ 数据库连接成功：{self.db_path}")

    def create_table(self):
        """创建空气质量数据表"""
        sql = """
        CREATE TABLE IF NOT EXISTS air_quality (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT NOT NULL DEFAULT 'device_001',
            record_time TEXT NOT NULL,
            pm25 REAL,
            pm10 REAL,
            temperature REAL,
            humidity REAL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            UNIQUE(device_id, record_time)
        );
        """
        self.cursor.execute(sql)
        self.conn.commit()
        print("✅ 数据表创建成功")

    def close(self):
        """关闭数据库连接"""
        if self.conn:
            self.conn.close()
            print("🔌 数据库连接已关闭")

# 使用示例
if __name__ == "__main__":
    db = AirQualityDB("air_quality.db")
    db.connect()
    db.create_table()
    db.close()

运行这段代码，你会在当前目录下看到 air_quality.db 文件。用 DB Browser for SQLite 打开看看，表结构一目了然。

注意：我曾经犯过一个低级错误——每次运行程序都重新建表，结果之前存的数据全没了。所以一定要用 IF NOT EXISTS，确保表已存在时不会覆盖。

4.5 插入数据：从清洗到入库

清洗完的数据怎么插进去？我一般这样写：

def insert_record(self, device_id, record_time, pm25, pm10, temp, humidity):
    """插入一条空气质量记录"""
    sql = """
    INSERT OR IGNORE INTO air_quality 
    (device_id, record_time, pm25, pm10, temperature, humidity)
    VALUES (?, ?, ?, ?, ?, ?)
    """
    try:
        self.cursor.execute(sql, (device_id, record_time, pm25, pm10, temp, humidity))
        self.conn.commit()
        return True
    except Exception as e:
        print(f"❌ 插入失败：{e}")
        return False

# 批量插入更高效
def insert_batch(self, records):
    """批量插入，records 是列表，每个元素是元组"""
    sql = """
    INSERT OR IGNORE INTO air_quality 
    (device_id, record_time, pm25, pm10, temperature, humidity)
    VALUES (?, ?, ?, ?, ?, ?)
    """
    try:
        self.cursor.executemany(sql, records)
        self.conn.commit()
        print(f"✅ 批量插入 {len(records)} 条记录")
        return True
    except Exception as e:
        print(f"❌ 批量插入失败：{e}")
        return False

注意 INSERT OR IGNORE 这个写法。如果遇到重复的 (device_id, record_time)，它不会报错，而是直接跳过。这在重复采集时特别有用。

4.6 验证数据是否存进去了

存完数据，总得看一眼吧？写个简单的查询：

def query_recent(self, limit=10):
    """查询最近 N 条记录"""
    sql = "SELECT * FROM air_quality ORDER BY record_time DESC LIMIT ?"
    self.cursor.execute(sql, (limit,))
    rows = self.cursor.fetchall()
    for row in rows:
        print(row)
    return rows

运行一下，你应该能看到类似这样的输出：

(1, 'device_001', '2024-01-15 08:30:00', 35.2, 68.5, 22.1, 45.3, '2024-01-15 08:30:05')
(2, 'device_001', '2024-01-15 08:31:00', 36.8, 70.1, 22.3, 44.9, '2024-01-15 08:31:05')
...

4.7 避坑指南

做数据存储这块，有几个坑我替你们踩过了：

常见问题 后果 解决方案
忘记 commit 数据没写进去，程序退出就丢了 每次 insert 后记得 conn.commit()
主键设计不合理 重复数据一堆，查起来头疼 用 UNIQUE 约束防重复
时间存成字符串但格式不统一 排序、过滤全乱套 统一用 ISO 格式：YYYY-MM-DD HH:MM:SS
没有关闭连接 数据库文件被锁，其他程序打不开 用 with 语句或显式 close()

我曾经在一个项目里，因为忘记 commit，跑了整整一天的数据全没了。从那以后，我每次写完 insert 都会下意识加一句 commit，都快成肌肉记忆了。

4.8 小结

这一章我们做了三件事：

  • 设计了合理的表结构，预留了扩展空间
  • 用 Python 的 sqlite3 模块连接并建表
  • 实现了单条和批量插入，以及简单的查询验证

数据存进去了，下一步就是怎么把这些数据可视化出来，让人一眼看懂空气质量的变化趋势。下一章我们聊聊数据可视化的事。

课后练习：试着在表里加一个 city 字段，然后插入几条不同城市的数据，看看能不能按城市分组查询。

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