一、问题背景Excel存不了这么多数据了项目做了一年后数据量开始失控。数据量- Lot记录10万条- Wafer测量数据250万条- 设备日志500万条- 良率数据1万条用Excel存的问题1. Excel每个sheet只有104万行——不够2. 打开100MB的Excel文件要2分钟3. 查询一条数据要CtrlF找半天4. 多人同时编辑会冲突解决方案用SQLite。---二、技术原理SQLite基础2.1 为什么用SQLiteimport 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数据管理系统FAB数据管理系统SQLite版核心功能用SQLite管理Lot数据支持查询异常与统计摘要import sqlite3import pandas as pdfrom typing import Dictimport logginglogging.basicConfig(levellogging.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_modeWAL)self._create_tables()def _create_tables(self):建表——为什么TEXT做主键Lot ID如FAB-001是业务标识不用自增INTcursor 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 Trueexcept sqlite3.IntegrityError:logger.warning(fLot {lot_data[lot_id]} 已存在跳过)return Falsedef query_abnormal_lots(self) - pd.DataFrame:查异常Lot——良率90或标记异常按良率升序排最差的在前return pd.read_sql_query(SELECT lot_id, process, wafer_count, thickness_avg, yield_rate, is_abnormalFROM lots WHERE is_abnormal 1 OR yield_rate 90ORDER 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_yieldFROM 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 0return {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_idFAB的人看数据时习惯按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资源专栏目录页可下载。
18.SQLite数据库操作：在本地管理FAB数据
发布时间：2026/6/26 12:10:23
一、问题背景Excel存不了这么多数据了项目做了一年后数据量开始失控。数据量- Lot记录10万条- Wafer测量数据250万条- 设备日志500万条- 良率数据1万条用Excel存的问题1. Excel每个sheet只有104万行——不够2. 打开100MB的Excel文件要2分钟3. 查询一条数据要CtrlF找半天4. 多人同时编辑会冲突解决方案用SQLite。---二、技术原理SQLite基础2.1 为什么用SQLiteimport 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数据管理系统FAB数据管理系统SQLite版核心功能用SQLite管理Lot数据支持查询异常与统计摘要import sqlite3import pandas as pdfrom typing import Dictimport logginglogging.basicConfig(levellogging.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_modeWAL)self._create_tables()def _create_tables(self):建表——为什么TEXT做主键Lot ID如FAB-001是业务标识不用自增INTcursor 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 Trueexcept sqlite3.IntegrityError:logger.warning(fLot {lot_data[lot_id]} 已存在跳过)return Falsedef query_abnormal_lots(self) - pd.DataFrame:查异常Lot——良率90或标记异常按良率升序排最差的在前return pd.read_sql_query(SELECT lot_id, process, wafer_count, thickness_avg, yield_rate, is_abnormalFROM lots WHERE is_abnormal 1 OR yield_rate 90ORDER 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_yieldFROM 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 0return {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_idFAB的人看数据时习惯按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资源专栏目录页可下载。
)
