SQLite数据库操作：在本地管理FAB数据
SQLite数据库操作：在本地管理FAB数据
一、问题背景：Excel存不了这么多数据了
项目做了一年后，数据量开始失控。
数据量：
- Lot记录：10万条
- Wafer测量数据：250万条
- 设备日志：500万条
- 良率数据：1万条
用Excel存的问题：
1. Excel每个sheet只有104万行——不够
2. 打开100MB的Excel文件要2分钟
3. 查询一条数据要Ctrl+F找半天
4. 多人同时编辑会冲突
解决方案：用SQLite。
---
二、技术原理：SQLite基础
2.1 为什么用SQLite？
```python
import sqlite3
连接到SQLite数据库（自动创建文件）
conn = sqlite3.connect('fab_data.db')
cursor = conn.cursor()
创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS lots (
lot_id TEXT PRIMARY KEY,
process TEXT NOT NULL,
wafer_count INTEGER,
thickness_avg REAL,
yield_rate REAL,
start_date TEXT,
is_abnormal INTEGER DEFAULT 0
)
''')
插入数据
cursor.execute('''
INSERT INTO lots (lot_id, process, wafer_count, thickness_avg, yield_rate, start_date)
VALUES (?, ?, ?, ?, ?, ?)
''', ('FAB-001', 'ETCH', 25, 1250.5, 96.5, '2026-01-15'))
conn.commit()
查询数据
cursor.execute("SELECT * FROM lots WHERE yield_rate < 90")
abnormal_lots = cursor.fetchall()
关闭连接
conn.close()
```
为什么用SQLite：
- 不需要安装数据库服务器
- 一个文件存所有数据
- 支持SQL查询
- 比Excel快100倍
---
三、实战案例：FAB数据管理系统
```python
"""
FAB数据管理系统（SQLite版）
核心功能：用SQLite管理Lot数据，支持查询异常与统计摘要
"""
import sqlite3
import pandas as pd
from typing import Dict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FABDatabase:
"""FAB数据库管理器——封装建表、插入、查询、统计四个核心操作"""
def __init__(self, db_path: str = 'fab_data.db'):
# 为什么用WAL模式？FAB多工序并行写入时，WAL允许读写并发，不锁库
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row # 按列名访问，比row[0]可读性好
self.conn.execute("PRAGMA journal_mode=WAL")
self._create_tables()
def _create_tables(self):
"""建表——为什么TEXT做主键？Lot ID如FAB-001是业务标识，不用自增INT"""
cursor = self.conn.cursor()
# lots表：每个Lot一行，is_abnormal用0/1便于SQL过滤
cursor.execute('''
CREATE TABLE IF NOT EXISTS lots (
lot_id TEXT PRIMARY KEY,
process TEXT NOT NULL,
wafer_count INTEGER,
thickness_avg REAL,
thickness_std REAL,
yield_rate REAL,
cycle_time INTEGER,
start_time TEXT,
status TEXT DEFAULT 'CREATED',
is_abnormal INTEGER DEFAULT 0
)''')
# 紧急查询靠索引——工序和异常标记是高频过滤条件
cursor.execute('CREATE INDEX IF NOT EXISTS idx_lot_process ON lots(process)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_lot_abnormal ON lots(is_abnormal)')
self.conn.commit()
def insert_lot(self, lot_data: Dict) -> bool:
"""插入Lot——为什么用?占位？防SQL注入，FAB数据有特殊字符时不会炸"""
try:
self.conn.execute('''
INSERT INTO lots (lot_id, process, wafer_count, thickness_avg,
thickness_std, yield_rate, cycle_time, start_time, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (lot_data['lot_id'], lot_data.get('process', ''),
lot_data.get('wafer_count', 0), lot_data.get('thickness_avg', 0),
lot_data.get('thickness_std', 0), lot_data.get('yield_rate', 100),
lot_data.get('cycle_time', 0), lot_data.get('start_time', ''),
lot_data.get('status', 'CREATED')))
self.conn.commit()
return True
except sqlite3.IntegrityError:
logger.warning(f"Lot {lot_data['lot_id']} 已存在，跳过")
return False
def query_abnormal_lots(self) -> pd.DataFrame:
"""查异常Lot——良率<90或标记异常，按良率升序排最差的在前"""
return pd.read_sql_query('''
SELECT lot_id, process, wafer_count, thickness_avg, yield_rate, is_abnormal
FROM lots WHERE is_abnormal = 1 OR yield_rate < 90
ORDER BY yield_rate ASC
''', self.conn)
def get_summary_statistics(self) -> Dict:
"""统计摘要——按工序分组是FAB分析最常用的视角"""
cursor = self.conn.cursor()
cursor.execute("SELECT COUNT(*) FROM lots")
total_lots = cursor.fetchone()[0]
cursor.execute('''
SELECT process, COUNT(*) as cnt, AVG(yield_rate) as avg_yield
FROM lots GROUP BY process
''')
process_stats = {r[0]: {'count': r[1], 'avg_yield': r[2]} for r in cursor.fetchall()}
cursor.execute("SELECT COUNT(*) FROM lots WHERE is_abnormal = 1")
abnormal_count = cursor.fetchone()[0]
cursor.execute("SELECT AVG(yield_rate) FROM lots")
avg_yield = cursor.fetchone()[0] or 0
return {'total_lots': total_lots, 'abnormal_count': abnormal_count,
'avg_yield': avg_yield, 'process_stats': process_stats}
def close(self):
if self.conn:
self.conn.close()
使用示例
if __name__ == '__main__':
db = FABDatabase('fab_data.db')
db.insert_lot({'lot_id': 'FAB-001', 'process': 'ETCH', 'wafer_count': 25,
'thickness_avg': 1250.5, 'thickness_std': 1.2, 'yield_rate': 96.5,
'cycle_time': 180, 'start_time': '2026-01-15T08:00'})
abnormal = db.query_abnormal_lots()
stats = db.get_summary_statistics()
print(f"总Lot数:{stats['total_lots']}, 平均良率:{stats['avg_yield']:.1f}%")
db.close()
```
---
四、数据库 vs Excel 对比
| 维度 | Excel | SQLite | 提升 |
|------|-------|--------|------|
| 最大行数 | 104万 | 理论无限制 | +∞ |
| 查询100万条 | 2分钟 | 0.1秒 | 1200倍 |
| 多表关联 | 困难 | SQL JOIN | 简单 |
| 文件大小 | 100MB+/慢 | 100MB+仍快 | +500% |
---
> 📦 专栏VIP资源包：包含本系列40篇全部可运行源码、示例数据集、自动化脚本工具包。在专栏主页点击「VIP资源」即可获取。
五、实施建议
从Excel迁移到SQLite不是一蹴而就的事。我在FAB里做了三个月才完全切换过来，踩了不少坑，总结几条实操建议。
数据库设计规范：表结构要围绕FAB业务实体来设计，不要照搬Excel的sheet名。我刚开始就是把每个Excel sheet直接转成一张表，结果查询时到处JOIN，性能很差。后来重新按业务实体（Lot、Wafer、Equipment）建表，每个实体一张主表，用外键关联。主键优先用业务ID（如lot_id），FAB的人看数据时习惯按ID说话，用自增整数主键反而要来回映射。索引要建在查询热点上——工序类型（process）、异常标记（is_abnormal）、时间范围（start_time），这三个是我日常查询最频繁的过滤条件。
从Excel迁移的步骤：第一步，先用pandas把Excel读进来做数据清洗，去掉重复行、空行、格式不一致的单元格。第二步，用`df.to_sql()`批量写入SQLite，这个方法比逐行INSERT快10倍以上。第三步，写几个验证查询，对比Excel和数据库的总行数、关键字段的均值，确保迁移没丢数据。我第一次迁移时发现少了200条记录，原因是Excel里有隐藏行，pandas默认不读隐藏行。所以迁移前一定要检查Excel的隐藏行和筛选状态。
备份策略：SQLite是单文件数据库，备份最简单的方式就是复制文件。我用Python写了个定时脚本，每天凌晨把`fab_data.db`复制到`fab_data_backup_YYYYMMDD.db`，保留最近30天的备份。更安全的方式是用SQLite的`.dump`命令导出SQL文本备份，文本备份不怕文件损坏。另外，WAL模式下备份前要先执行`PRAGMA wal_checkpoint(FULL)`，把WAL日志合并到主文件，否则备份可能不完整。
六、进阶方向
SQLite对于单机FAB数据分析足够了，但当数据量超过千万级或需要多人并发写入时，就该考虑升级了。
SQLAlchemy ORM：直接写SQL容易出错，字段名拼错、WHERE条件漏写都很常见。SQLAlchemy用Python类映射数据库表，`Lot.query.filter(Lot.yield_rate < 90).all()`比手写SQL直观得多，还能自动处理连接管理和事务。我后来把整个FAB数据分析项目都迁移到了SQLAlchemy，代码量减少了30%，维护效率显著提升。入门推荐从`flask-sqlalchemy`开始，文档齐全、示例丰富，半天就能上手。
PostgreSQL升级：当FAB数据需要多人实时写入、或者查询涉及复杂的窗口函数（如按工序计算滚动平均良率），SQLite就力不从心了。PostgreSQL支持并发写入、JSON字段（存设备参数这种半结构化数据）、窗口函数、物化视图。迁移成本不高——SQL语法基本兼容，pandas的`to_sql()`换个连接字符串就行。部署可以用Docker，一条命令拉起来：`docker run -d -p 5432:5432 postgres:15`。
数据仓库概念：FAB数据分析最终会走向OLAP方向——按工序、时间、设备做多维聚合。这时候可以引入数据仓库的分层设计：ODS层（原始数据）→DWD层（清洗后明细）→DWS层（按工序聚合的汇总表）→ADS层（分析结果）。每层一张表，查询时直接读ADS层，不用每次重新聚合。这个思路用SQLite也能实现，就是多建几张汇总表配合定时刷新脚本。但数据量大了之后，建议用ClickHouse或者DuckDB这类列式数据库做OLAP引擎，聚合查询速度比SQLite快10-50倍。
七、总结
SQLite是一个零配置的数据库。不需要安装服务器，一个文件搞定所有数据。对于FAB数据分析来说，足够用了。
下一篇预告：API数据采集——从MES系统自动拉取数据。
---
> 💬 你在实际工作中遇到过类似问题吗？欢迎在评论区聊聊你的经历，或者说说你最想用Python自动化的场景！
>
> 📚 专栏持续更新中，关注不迷路。觉得有用的话，收藏+点赞支持一下~ 👍
>
> 🔧 专栏配套工具包（含本篇完整可运行代码+示例数据）已上传为VIP资源，专栏目录页可下载。
