Python & SQLite为什么是小型工业数据库的最佳选择?

这篇文章写给那些在服务器上装了一堆中间件、配了半天连接池、结果发现数据量根本没到瓶颈的朋友。

🤔 你真的需要那么重的数据库吗？

工业现场的数据库需求，和互联网业务有本质区别——并发量不高，但可靠性要求极高；数据量不小，但查询模式极为固定；部署环境受限，但运维能力几乎为零。

SQLite在这个场景里，不是"将就用用"，而是真正的最优解。接下来，咱们好好掰扯一下为什么。

🏭 工业数据库的真实需求画像

很多人一提"工业数据库"，脑子里冒出来的是Oracle、InfluxDB、甚至Hadoop。但现实中，工业边缘端的需求往往是这样的：

• • 单机部署，没有网络，没有DBA
• • 写入频率稳定（1Hz ~ 100Hz），读取偶发
• • 数据保留周期明确（比如只保留最近30天）
• • 需要支持简单聚合查询（最大值、均值、趋势）
• • 断电恢复后必须数据完整

你拿这个需求去套MySQL或PostgreSQL——能用，但就像用卡车拉自行车，浪费且麻烦。

SQLite的设计哲学恰好契合这个场景：serverless、zero-configuration、self-contained。它不是玩具，NASA用它，Airbus用它，Android系统的联系人数据库也是它。

💡 SQLite的底层机制，你可能不知道这些

📦 WAL模式：写入性能的关键开关

默认情况下SQLite用的是DELETE日志模式，写入时会锁整个数据库文件。但开启**WAL（Write-Ahead Logging）**之后，读写可以并发——写入先追加到WAL文件，读取直接走主库快照。

这个开关，很多人从来没打开过。

import sqlite3conn = sqlite3.connect("industrial.db")# 开启WAL模式，写入性能提升3~5倍conn.execute("PRAGMA journal_mode=WAL;")# 关闭同步等待，适合非关键写入场景conn.execute("PRAGMA synchronous=NORMAL;")# 调大缓存页（默认2MB，改成32MB）conn.execute("PRAGMA cache_size=-32000;")conn.commit()

实测数据：同样的写入场景，开启WAL后吞吐量从约800条/秒提升到4000+条/秒，延迟从均值12ms降到2ms以内。这不是理论值，是我在i5工控机上跑出来的。

🗜️ 数据压缩与页大小优化

SQLite默认页大小是4096字节。对于时序数据，每条记录字段少、数值型居多，适当调大页大小可以减少I/O次数：

# 注意：page_size必须在建库前设置，建库后无法修改conn = sqlite3.connect("new_database.db")conn.execute("PRAGMA page_size=8192;")conn.execute("VACUUM;")  # 重建数据库以应用新页大小

对于字符串密集型数据（比如报警日志），可以考虑在Python层做压缩后存BLOB，查询时解压——但这是后话，先把基础调优做扎实。

🔧 方案一：高频写入的正确姿势

场景描述

PLC每100ms上报一次数据，Python脚本接收后写入本地SQLite。如果每条数据都单独commit，性能会很惨。

错误做法（别这么写）

# ❌ 这是新手常见写法，每条记录都提交一次事务for record in data_stream:    conn.execute("INSERT INTO sensor_data VALUES (?, ?, ?)", record)    conn.commit()  # 每次都刷盘，慢得要命

正确做法：批量提交 + 异常保护

import sqlite3  import time  from collections import deque  from threading import Thread, Lock  from datetime import datetime, timedelta  import statistics  classIndustrialDataWriter:  def__init__(self, db_path: str, batch_size: int = 500, flush_interval: float = 2.0):  self.db_path = db_path  self.batch_size = batch_size  self.flush_interval = flush_interval  self.buffer = deque()  self.lock = Lock()  self.stats = {  'total_writes': 0,  'total_flushes': 0,  'write_errors': 0        }  self._running = Trueself._init_db()  # 启动后台刷盘线程  self._flush_thread = Thread(target=self._auto_flush, daemon=True)  self._flush_thread.start()  def_init_db(self):  """初始化数据库和优化配置"""        conn = sqlite3.connect(self.db_path)          conn.execute("PRAGMA journal_mode=WAL;")  # 预写式日志，提高并发性能          conn.execute("PRAGMA synchronous=NORMAL;")  # 平衡性能和安全性          conn.execute("PRAGMA cache_size=-32000;")  # 32MB缓存          conn.execute("""              CREATE TABLE IF NOT EXISTS sensor_data (                id        INTEGER PRIMARY KEY AUTOINCREMENT,                ts        INTEGER NOT NULL,   -- Unix时间戳（毫秒）                  device_id TEXT    NOT NULL,                value     REAL    NOT NULL,                quality   INTEGER DEFAULT 1  -- 数据质量标志              )        """)  # 时间戳索引，查询时间范围必备          conn.execute("CREATE INDEX IF NOT EXISTS idx_ts ON sensor_data(ts);")          conn.execute("CREATE INDEX IF NOT EXISTS idx_device ON sensor_data(device_id, ts);")          conn.commit()          conn.close()  defwrite(self, device_id: str, value: float, quality: int = 1):  """非阻塞写入，数据先进缓冲区"""        ts = int(time.time() * 1000)  withself.lock:  self.buffer.append((ts, device_id, value, quality))  self.stats['total_writes'] += 1# 缓冲区满时触发刷盘  iflen(self.buffer) >= self.batch_size:  self._flush()  def_flush(self):  """将缓冲区数据批量写入数据库"""ifnotself.buffer:  return        batch = []  whileself.buffer:              batch.append(self.buffer.popleft())  try:              conn = sqlite3.connect(self.db_path)              conn.executemany(  "INSERT INTO sensor_data(ts, device_id, value, quality) VALUES (?,?,?,?)",                  batch              )              conn.commit()              conn.close()  self.stats['total_flushes'] += 1print(f"[✓ 刷盘成功] 写入 {len(batch)} 条记录")  except sqlite3.Error as e:  # 写入失败时把数据塞回缓冲区，不丢数据  print(f"[✗ 写入失败] {e}，数据回滚至缓冲区")  self.stats['write_errors'] += 1withself.lock:  # 将批次数据按逆序放回队列前端  self.buffer.extendleft(reversed(batch))  def_auto_flush(self):  """定时刷盘，保证数据及时落地"""whileself._running:              time.sleep(self.flush_interval)  withself.lock:  ifself.buffer:  self._flush()  defforce_flush(self):  """主动刷盘"""withself.lock:  self._flush()  defget_buffer_size(self):  """获取当前缓冲区大小"""withself.lock:  returnlen(self.buffer)  defget_stats(self):  """获取统计信息"""withself.lock:  returnself.stats.copy()  defclose(self):  """优雅关闭"""self._running = False        time.sleep(self.flush_interval + 0.1)  withself.lock:  self._flush()  # 最后一次刷盘  print("[关闭] IndustrialDataWriter 已完成最后的数据刷盘")  classDataQuery:  """数据查询工具类"""def__init__(self, db_path: str):  self.db_path = db_path  defquery_by_time_range(self, device_id: str, start_ms: int, end_ms: int):  """按时间范围查询"""        conn = sqlite3.connect(self.db_path)          cursor = conn.execute(  """SELECT ts, value, quality FROM sensor_data                  WHERE device_id = ? AND ts BETWEEN ? AND ?                 ORDER BY ts""",              (device_id, start_ms, end_ms)          )          results = cursor.fetchall()          conn.close()  return results  defget_latest_value(self, device_id: str):  """获取最新数据"""        conn = sqlite3.connect(self.db_path)          cursor = conn.execute(  """SELECT ts, value, quality FROM sensor_data                  WHERE device_id = ?                  ORDER BY ts DESC LIMIT 1""",              (device_id,)          )          result = cursor.fetchone()          conn.close()  return result  defget_statistics(self, device_id: str, start_ms: int, end_ms: int):  """统计分析"""        data = self.query_by_time_range(device_id, start_ms, end_ms)  ifnot data:  returnNone        values = [row[1] for row in data]  return {  'count': len(values),  'min': min(values),  'max': max(values),  'avg': statistics.mean(values),  'stdev': statistics.stdev(values) iflen(values) > 1else0,  'median': statistics.median(values)          }  defdelete_old_data(self, days: int):  """删除旧数据"""        cutoff_ms = int((time.time() - days * 86400) * 1000)          conn = sqlite3.connect(self.db_path)          cursor = conn.execute("DELETE FROM sensor_data WHERE ts < ?", (cutoff_ms,))          deleted = cursor.rowcount          conn.commit()          conn.close()  return deleted  # ===================== 使用示例 =====================  if __name__ == "__main__":  import random  print("=" * 60)  print("工业级数据写入系统 - 演示")  print("=" * 60)  # 初始化写入器      writer = IndustrialDataWriter("sensor.db", batch_size=100, flush_interval=1.0)  print("\n[1] 模拟数据写入...")      devices = ["PUMP-001", "TEMP-002", "PRESSURE-003", "FLOW-004"]  # 写入1000条数据  for i inrange(1000):          device_id = random.choice(devices)          value = random.gauss(100, 15)  # 均值100，标准差15          quality = random.choice([1, 1, 1, 0])  # 75%高质量数据          writer.write(device_id, value, quality)  if (i + 1) % 250 == 0:  print(f"  → 已写入 {i + 1} 条数据，缓冲区: {writer.get_buffer_size()}")              time.sleep(0.1)  # 等待刷盘完成  print("\n[2] 等待缓冲数据刷盘...")      time.sleep(2)      writer.force_flush()  # 查询统计  print("\n[3] 数据查询和统计...")      query = DataQuery("sensor.db")      now_ms = int(time.time() * 1000)      one_hour_ago = now_ms - 3600 * 1000for device_id in devices:          stats = query.get_statistics(device_id, one_hour_ago, now_ms)  if stats:  print(f"\n  设备: {device_id}")  print(f"    数据量: {stats['count']}")  print(f"    范围: [{stats['min']:.2f}, {stats['max']:.2f}]")  print(f"    平均: {stats['avg']:.2f}")  print(f"    标准差: {stats['stdev']:.2f}")  print(f"    中位数: {stats['median']:.2f}")  # 统计信息  print("\n[4] 系统统计信息...")      stats = writer.get_stats()  print(f"  总写入次数: {stats['total_writes']}")  print(f"  总刷盘次数: {stats['total_flushes']}")  print(f"  写入错误: {stats['write_errors']}")  # 优雅关闭  print("\n[5] 优雅关闭...")      writer.close()  print("\n✓ 演示完成")

踩坑预警：deque不是线程安全的，上面代码用Lock保护了关键区域。如果你把Lock去掉，在高并发下会出现数据乱序甚至丢失——这个错误我在代码审查里见过不止三次。

🔍 方案二：时序查询的性能陷阱与优化

场景描述

查询某设备过去24小时的平均值、最大值，并按小时聚合展示趋势图。

索引设计是关键

import sqlite3  import time  from datetime import datetime, timedelta  from typing importList, Dict, Tuplefrom enum import Enum  import statistics  import json  classTimeGranularity(Enum):  """时间粒度"""    MINUTE = ("分钟", 60 * 1000)      HOUR = ("小时", 3600 * 1000)      DAY = ("天", 24 * 3600 * 1000)      WEEK = ("周", 7 * 24 * 3600 * 1000)  classTimeSeriesAggregator:  """时间序列数据聚合查询引擎"""def__init__(self, db_path: str):  self.db_path = db_path  self._register_custom_functions()  def_register_custom_functions(self):  """注册SQLite自定义函数"""        conn = sqlite3.connect(self.db_path)  # 标准差计算（聚合函数）  classStddevAggregate:  def__init__(self):  self.values = []  defstep(self, value):  if value isnotNone:  self.values.append(value)  deffinalize(self):  iflen(self.values) <= 1:  return0.0return statistics.stdev(self.values)          conn.create_aggregate("STDDEV", 1, StddevAggregate)          conn.close()  defquery_hourly_stats(self, device_id: str, hours: int = 24) -> List[Dict]:  """          按小时聚合查询，返回每小时的均值和峰值          注意：ts字段存的是毫秒时间戳          """        end_ts = int(time.time() * 1000)          start_ts = end_ts - hours * 3600 * 1000        conn = sqlite3.connect(self.db_path)  self._register_custom_functions()          sql = """              SELECT                -- 将毫秒时间戳转换为小时级别的桶                  (ts / 3600000) * 3600000  AS hour_bucket,                AVG(value)                AS avg_val,                MAX(value)                AS max_val,                MIN(value)                AS min_val,                COUNT(*)                  AS sample_count            FROM sensor_data            WHERE device_id = ?              AND ts BETWEEN ? AND ?              AND quality = 1            GROUP BY hour_bucket            ORDER BY hour_bucket ASC        """        cursor = conn.execute(sql, (device_id, start_ts, end_ts))          rows = cursor.fetchall()          conn.close()  return [              {                "hour": row[0],  "hour_readable": self._ms_to_datetime(row[0]),  "avg": round(row[1], 3) if row[1] elseNone,  "max": round(row[2], 3) if row[2] elseNone,  "min": round(row[3], 3) if row[3] elseNone,  "count": row[4],  "range": round(row[2] - row[3], 3) if row[2] and row[3] elseNone            }  for row in rows          ]  defquery_with_granularity(self, device_id: str, granularity: TimeGranularity,                                 periods: int = 24) -> List[Dict]:  """          按指定粒度聚合查询（修复版本 - 移除SQLite不支持的函数）          Args:            device_id: 设备ID              granularity: 时间粒度（分钟/小时/天/周）              periods: 时间周期数          """        bucket_ms = granularity.value[1]          end_ts = int(time.time() * 1000)          start_ts = end_ts - periods * bucket_ms          conn = sqlite3.connect(self.db_path)  self._register_custom_functions()  # 【修复】移除 STDDEV_POP，改用自定义 STDDEV 或Python计算          sql = """              SELECT                (ts / ?) * ? AS time_bucket,                AVG(value) AS avg_val,                MAX(value) AS max_val,                MIN(value) AS min_val,                COUNT(*) AS sample_count,                SUM(CASE WHEN quality = 0 THEN 1 ELSE 0 END) AS bad_quality_count            FROM sensor_data            WHERE device_id = ?              AND ts BETWEEN ? AND ?            GROUP BY time_bucket            ORDER BY time_bucket ASC        """        cursor = conn.execute(sql, (bucket_ms, bucket_ms, device_id, start_ts, end_ts))          rows = cursor.fetchall()          conn.close()  # 获取每个时间桶的所有原始值，计算标准差          results = []  for row in rows:              time_bucket = row[0]              avg_val = row[1]              max_val = row[2]              min_val = row[3]              sample_count = row[4]              bad_quality = row[5]  if avg_val isnotNone:  # 计算该时间桶的标准差                  stddev = self._calculate_stddev_for_bucket(device_id, time_bucket, bucket_ms)                  results.append({  "bucket_ms": time_bucket,  "bucket_readable": self._ms_to_datetime(time_bucket),  "avg": round(avg_val, 3),  "max": round(max_val, 3),  "min": round(min_val, 3),  "stddev": round(stddev, 3),  "total_samples": sample_count,  "bad_quality_samples": bad_quality,  "data_quality": round((sample_count - bad_quality) / sample_count * 100,  2) if sample_count > 0else0                })  return results  def_calculate_stddev_for_bucket(self, device_id: str, bucket_ms: int, bucket_size_ms: int) -> float:  """计算指定时间桶的标准差"""        end_bucket = bucket_ms + bucket_size_ms          conn = sqlite3.connect(self.db_path)          cursor = conn.execute(  "SELECT value FROM sensor_data WHERE device_id = ? AND ts >= ? AND ts < ?",              (device_id, bucket_ms, end_bucket)          )          values = [row[0] for row in cursor.fetchall()]          conn.close()  iflen(values) <= 1:  return0.0return statistics.stdev(values)  defquery_with_anomaly_detection(self, device_id: str, hours: int = 24,                                       threshold_sigma: float = 2.5) -> Dict:  """          带异常检测的聚合查询          Args:            device_id: 设备ID              hours: 查询时长（小时）              threshold_sigma: 异常检测阈值（标准差倍数）          """        stats = self.query_hourly_stats(device_id, hours)  ifnot stats:  return {"error": "无数据"}          avg_values = [s["avg"] for s in stats if s["avg"] isnotNone]  ifnot avg_values:  return {"error": "无有效数据"}  # 计算统计指标          mean_val = statistics.mean(avg_values)          stdev_val = statistics.stdev(avg_values) iflen(avg_values) > 1else0        upper_bound = mean_val + threshold_sigma * stdev_val          lower_bound = mean_val - threshold_sigma * stdev_val  # 标记异常点  for item in stats:  if item["avg"] isnotNone:                  item["is_anomaly"] = item["avg"] > upper_bound or item["avg"] < lower_bound                  item["z_score"] = (item["avg"] - mean_val) / stdev_val if stdev_val > 0else0else:                  item["is_anomaly"] = False                item["z_score"] = 0        anomalies = [item for item in stats if item["is_anomaly"]]  return {  "device_id": device_id,  "hours": hours,  "total_buckets": len(stats),  "anomaly_count": len(anomalies),  "anomaly_rate": round(len(anomalies) / len(stats) * 100, 2) if stats else0,  "statistics": {  "mean": round(mean_val, 3),  "stdev": round(stdev_val, 3),  "upper_bound": round(upper_bound, 3),  "lower_bound": round(lower_bound, 3),  "min": round(min(avg_values), 3),  "max": round(max(avg_values), 3)              },  "anomalies": anomalies[:10]  # 返回前10个异常          }  defquery_trend_analysis(self, device_id: str, hours: int = 24) -> Dict:  """          趋势分析：计算增长率、加速度等          """        stats = self.query_hourly_stats(device_id, hours)  iflen(stats) < 2:  return {"error": "数据不足"}          avg_values = [s["avg"] for s in stats if s["avg"] isnotNone]  iflen(avg_values) < 2:  return {"error": "有效数据不足"}  # 计算逐小时变化率          changes = []  for i inrange(1, len(avg_values)):  if avg_values[i - 1] != 0:                  change_rate = (avg_values[i] - avg_values[i - 1]) / abs(avg_values[i - 1]) * 100else:                  change_rate = 0            changes.append(change_rate)  # 线性回归（简单）          n = len(avg_values)          x = list(range(n))          y = avg_values          x_mean = statistics.mean(x)          y_mean = statistics.mean(y)          numerator = sum((x[i] - x_mean) * (y[i] - y_mean) for i inrange(n))          denominator = sum((x[i] - x_mean) ** 2for i inrange(n))          slope = numerator / denominator if denominator != 0else0        intercept = y_mean - slope * x_mean  return {  "device_id": device_id,  "hours": hours,  "trend": {  "slope": round(slope, 4),  # 单位：值/小时  "intercept": round(intercept, 3),  "direction": "↑ 上升"if slope > 0.01else"↓ 下降"if slope < -0.01else"→ 平稳",  "change_rate_per_hour": round(slope / y_mean * 100, 2) if y_mean != 0else0            },  "volatility": {  "avg_change_rate": round(statistics.mean(changes), 3),  "max_change_rate": round(max(changes), 3),  "min_change_rate": round(min(changes), 3),  "volatility_stddev": round(statistics.stdev(changes), 3) iflen(changes) > 1else0            },  "first_value": round(avg_values[0], 3),  "last_value": round(avg_values[-1], 3),  "total_change": round(avg_values[-1] - avg_values[0], 3),  "total_change_percent": round((avg_values[-1] - avg_values[0]) / abs(avg_values[0]) * 100, 2) if avg_values[  0] != 0else0        }  defquery_compare_devices(self, device_ids: List[str], hours: int = 24) -> Dict:  """          多设备对比          """        comparison = {}  for device_id in device_ids:              stats = self.query_hourly_stats(device_id, hours)  if stats:                  avg_values = [s["avg"] for s in stats if s["avg"] isnotNone]  if avg_values:                      comparison[device_id] = {  "count": len(avg_values),  "avg_value": round(statistics.mean(avg_values), 3),  "max_value": round(max(avg_values), 3),  "min_value": round(min(avg_values), 3),  "stddev": round(statistics.stdev(avg_values), 3) iflen(avg_values) > 1else0,  "cv": round(statistics.stdev(avg_values) / statistics.mean(avg_values) * 100,  2) if statistics.mean(avg_values) != 0else0                    }  return comparison  defexport_to_csv(self, device_id: str, hours: int, output_file: str):  """          导出为CSV格式          """        stats = self.query_hourly_stats(device_id, hours)  ifnot stats:  print("无数据可导出")  returnimport csv  withopen(output_file, 'w', newline='', encoding='utf-8') as f:              writer = csv.DictWriter(f, fieldnames=stats[0].keys())              writer.writeheader()              writer.writerows(stats)  print(f"✓ 已导出到 {output_file}")      @staticmethod  def_ms_to_datetime(ms: int) -> str:  """毫秒时间戳转可读日期时间"""return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d %H:%M:%S")  # ===================== 使用示例 =====================  if __name__ == "__main__":  import random  # 生成测试数据  print("=" * 70)  print("时间序列数据聚合查询系统 - 演示（修复版本）")  print("=" * 70)      conn = sqlite3.connect("sensor.db")  # 删除旧表  try:          conn.execute("DROP TABLE IF EXISTS sensor_data")  except:  pass# 创建新表      conn.execute("PRAGMA journal_mode=WAL;")      conn.execute("PRAGMA synchronous=NORMAL;")      conn.execute("""          CREATE TABLE IF NOT EXISTS sensor_data (            id        INTEGER PRIMARY KEY AUTOINCREMENT,            ts        INTEGER NOT NULL,            device_id TEXT    NOT NULL,            value     REAL    NOT NULL,            quality   INTEGER DEFAULT 1        )    """)      conn.execute("CREATE INDEX IF NOT EXISTS idx_device_ts ON sensor_data(device_id, ts);")  # 生成5天的模拟数据  print("\n[1] 生成模拟数据...")      devices = ["PUMP-001", "TEMP-002", "PRESSURE-003"]      base_ts = int(time.time() * 1000) - 5 * 24 * 3600 * 1000for device_id in devices:          base_value = {"PUMP-001": 80, "TEMP-002": 25, "PRESSURE-003": 10}[device_id]  for i inrange(5 * 24 * 60):  # 5天×24小时×60分钟              ts = base_ts + i * 60 * 1000# 添加正弦波趋势              value = base_value + 10 * (i % 1440) / 1440 + random.gauss(0, 2)              quality = 1if random.random() > 0.05else0            conn.execute(  "INSERT INTO sensor_data(ts, device_id, value, quality) VALUES (?,?,?,?)",                  (ts, device_id, value, quality)              )      conn.commit()      conn.close()  print(f"  ✓ 已生成 {len(devices) * 5 * 24 * 60} 条数据")  # 初始化聚合器      agg = TimeSeriesAggregator("sensor.db")  # 1. 按小时聚合  print("\n[2] 小时级别聚合（最近24小时）...")      hourly_stats = agg.query_hourly_stats("PUMP-001", hours=24)  for stat in hourly_stats[:5]:  # 显示前5条  print(  f"  {stat['hour_readable']}: avg={stat['avg']}, max={stat['max']}, min={stat['min']}, count={stat['count']}")  # 2. 多粒度查询【修复】  print("\n[3] 不同粒度聚合（天级别，最近7天）...")  try:          daily_stats = agg.query_with_granularity("TEMP-002", TimeGranularity.DAY, periods=7)  for stat in daily_stats:  print(  f"  {stat['bucket_readable']}: avg={stat['avg']}, stddev={stat['stddev']}, 数据质量={stat['data_quality']}%")  except Exception as e:  print(f"  ✗ 查询失败: {e}")  # 3. 异常检测  print("\n[4] 异常检测分析...")      anomaly_result = agg.query_with_anomaly_detection("PRESSURE-003", hours=24, threshold_sigma=2.0)  if"error"notin anomaly_result:  print(  f"  异常数据点: {anomaly_result['anomaly_count']}/{anomaly_result['total_buckets']} ({anomaly_result['anomaly_rate']}%)")  print(  f"  统计指标: 均值={anomaly_result['statistics']['mean']}, 标准差={anomaly_result['statistics']['stdev']}")  print(  f"  正常范围: [{anomaly_result['statistics']['lower_bound']}, {anomaly_result['statistics']['upper_bound']}]")  else:  print(f"  {anomaly_result['error']}")  # 4. 趋势分析  print("\n[5] 趋势分析...")      trend = agg.query_trend_analysis("PUMP-001", hours=24)  if"error"notin trend:  print(f"  趋势方向: {trend['trend']['direction']}")  print(f"  增长率: {trend['trend']['change_rate_per_hour']}% / 小时")  print(f"  总变化: {trend['total_change']} ({trend['total_change_percent']}%)")  print(f"  波动度标准差: {trend['volatility']['volatility_stddev']}")  else:  print(f"  {trend['error']}")  # 5. 多设备对比  print("\n[6] 多设备对比...")      comparison = agg.query_compare_devices(devices, hours=24)  for device_id, metrics in comparison.items():  print(f"  {device_id}:")  print(f"    平均值: {metrics['avg_value']}, 范围: [{metrics['min_value']}, {metrics['max_value']}]")  print(f"    变异系数: {metrics['cv']}% (稳定性指标)")  # 6. 导出CSV  print("\n[7] 导出数据...")      agg.export_to_csv("PUMP-001", 24, "pump_data.csv")  print("\n✓ 演示完成")

性能对比：在300万条记录的表上，没有索引时这个查询需要约4.2秒；建立(device_id, ts)复合索引后，降到38毫秒。差了两个数量级——这就是索引的价值。

🧹 方案三：数据老化清理，别让数据库无限膨胀

工业现场一个常见问题：数据库文件越来越大，最后把存储空间撑爆。定期清理 + VACUUM 是必须要做的事情。

import sqlite3import timeimport loggingdefcleanup_old_data(db_path: str, retention_days: int = 30):"""    清理超过保留期的历史数据    建议在凌晨低负载时段执行    """    cutoff_ts = int((time.time() - retention_days * 86400) * 1000)    conn = sqlite3.connect(db_path)try:# 分批删除，避免长事务锁表        batch_size = 10000        total_deleted = 0whileTrue:            cursor = conn.execute("""                DELETE FROM sensor_data                WHERE id IN (                    SELECT id FROM sensor_data                    WHERE ts < ?                    LIMIT ?                )                """,                (cutoff_ts, batch_size)            )            conn.commit()            deleted = cursor.rowcount            total_deleted += deletedif deleted < batch_size:break# 没有更多旧数据了            time.sleep(0.1)  # 给其他操作留点喘息空间        logging.info(f"清理完成，共删除 {total_deleted} 条记录")# 回收磁盘空间（注意：VACUUM会重建数据库文件，比较耗时）        conn.execute("VACUUM;")        conn.close()except Exception as e:        logging.error(f"清理失败: {e}")        conn.close()raise

扩展建议：如果数据量特别大（单表超过5000万行），可以考虑按月分表——sensor_data_202601、sensor_data_202602这样，清理时直接DROP TABLE，比DELETE快几十倍，也不需要VACUUM。

⚠️ 那些让人踩坑的常见误解

误解一："SQLite不支持并发，多线程会崩"

实际上，SQLite在WAL模式下支持多读一写。Python的sqlite3模块默认check_same_thread=True，改成False并配合连接池就能多线程使用。但要注意：每个线程用独立的连接对象，不要共享同一个conn。

误解二："数据量超过几百万就要换数据库"

我见过单文件20GB、记录数过亿的SQLite库，查询依然很快——前提是索引设计合理，查询语句写得对。盲目换库，往往是因为SQL写烂了，不是SQLite的锅。

误解三："SQLite没有权限控制，不安全"

工业边缘端的安全威胁模型和互联网完全不同。物理隔离的设备上，文件系统权限 + 应用层加密（SQLCipher）完全够用。非要在工控机上搭一套RBAC，才是真正的过度设计。

🚀 进阶：用Python封装一个工业级数据访问层

把上面所有内容整合起来，一个可以直接复用的模板：