影刀RPA 数据库查询优化索引与分页署名林焱什么情况用你的RPA流程跑着跑着越来越慢查个订单从秒回变成卡30秒数据库里攒了上百万条数据COUNT一下都要十几秒这不是数据库不行是你没建索引、不会分页。查询优化不是DBA才需要学的技能。RPA开发者的SQL写得好不好直接决定流程是3秒跑完还是3分钟卡死。典型场景数据库表超过10万行查询明显变慢需要按多个条件组合查询如店铺日期状态RPA流程需要分批处理大量数据每次取1000条查询结果需要实时返回不能让用户等太久怎么做1. 建索引importsqlite3 db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 单列索引加速按单列查询cursor.execute(CREATE INDEX IF NOT EXISTS idx_shop ON products(shop))cursor.execute(CREATE INDEX IF NOT EXISTS idx_price ON products(price))cursor.execute(CREATE INDEX IF NOT EXISTS idx_crawl_time ON products(crawl_time))# 联合索引加速多列组合查询最左前缀原则cursor.execute(CREATE INDEX IF NOT EXISTS idx_shop_price ON products(shop, price))cursor.execute(CREATE INDEX IF NOT EXISTS idx_tags_time ON products(tags, crawl_time))[video(video-nc9GGDqh-1784264853162)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限)]conn.commit()print(索引创建完成)# 验证索引是否被使用cursor.execute(EXPLAIN QUERY PLAN SELECT * FROM products WHERE shop 京东 AND price 100)plancursor.fetchall()print(查询计划)forrowinplan:print(f{row})# 如果出现 USING INDEX idx_shop_price 说明索引生效# 如果出现 SCAN TABLE 说明全表扫描索引没生效conn.close()2. 分页查询importsqlite3 db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 基础分页LIMIT OFFSETpage1page_size100cursor.execute(SELECT COUNT(*) FROM products)totalcursor.fetchone()[0]total_pages(totalpage_size-1)//page_sizeprint(f总记录{total}总页数{total_pages})# 取第1页cursor.execute(SELECT * FROM products LIMIT ? OFFSET ?,(page_size,(page-1)*page_size))page_datacursor.fetchall()print(f第{page}页{len(page_data)}条)conn.close()影刀操作配合用【循环-次数】组件循环页数每页取100条数据 → 【执行Python代码】处理 → 【写入Excel文件】追加写入。3. 游标分页大数据量推荐importsqlite3 db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 传统分页的问题OFFSET 100000 LIMIT 100 要先扫描前10万条很慢# 游标分页记住上一页最后一条的ID下一页从该ID之后取page_size1000last_id0total_processed0whileTrue:cursor.execute(SELECT id, product_name, price FROM products WHERE id ? ORDER BY id LIMIT ?,(last_id,page_size))batchcursor.fetchall()ifnotbatch:break# 没有更多数据# 处理这批数据forrowinbatch:# 业务处理...passlast_idbatch[-1][0]# 记住最后一条的IDtotal_processedlen(batch)print(f已处理{total_processed})print(f全部完成共处理{total_processed}条)conn.close()4. 只查需要的列importsqlite3 db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 错误SELECT * 读取所有列包括不需要的大字段# cursor.execute(SELECT * FROM products WHERE shop 京东)# 正确只查需要的列cursor.execute(SELECT product_name, price FROM products WHERE shop 京东)resultscursor.fetchall()conn.close()5. 查询优化检查工具importsqlite3importtime db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()defquery_with_timing(sql,paramsNone,label):带计时和执行计划分析的查询# 先看执行计划cursor.execute(fEXPLAIN QUERY PLAN{sql},paramsor())plancursor.fetchall()# 执行查询starttime.time()cursor.execute(sql,paramsor())resultscursor.fetchall()elapsedtime.time()-start# 输出分析print(f\n[{label}])print(f SQL:{sql[:80]}...)print(f 计划:{plan[-1][3]ifplanelseN/A})print(f 结果:{len(results)}条, 耗时{elapsed:.3f}秒)returnresults# 对比有无索引的查询速度query_with_timing(SELECT * FROM products WHERE shop 京东,label按店铺查询)query_with_timing(SELECT * FROM products WHERE shop 京东 AND price 100,label店铺价格查询)query_with_timing(SELECT * FROM products WHERE product_name LIKE %耳机%,label模糊查询)conn.close()有什么坑坑1LIKE ‘%关键词’ 用不上索引# 前缀匹配可以用索引cursor.execute(SELECT * FROM products WHERE product_name LIKE 耳机%)# 快# 左模糊或全模糊用不上索引cursor.execute(SELECT * FROM products WHERE product_name LIKE %耳机%)# 慢全表扫描如果必须做全模糊搜索两个方案方案1在Python里做过滤数据量小时方案2用SQLite的FTS全文搜索# 创建FTS虚拟表cursor.execute(CREATE VIRTUAL TABLE IF NOT EXISTS products_fts USING fts5(product_name, contentproducts, content_rowidid))# 填充数据cursor.execute(INSERT INTO products_fts(product_name) SELECT product_name FROM products)# 全文搜索快cursor.execute(SELECT * FROM products WHERE id IN (SELECT rowid FROM products_fts WHERE product_name MATCH 耳机))坑2OFFSET越大越慢LIMIT 100 OFFSET 100000需要扫描前100100条记录然后丢弃前10万条。数据量大时第1000页比第1页慢100倍。游标分页WHERE id last_id不受页数影响每页都是常数时间。但游标分页不支持跳页——只能下一页不能跳到第50页。如果你的RPA流程是顺序处理的用游标分页如果需要随机跳页用OFFSET但限制最大页数。坑3联合索引的最左前缀原则建了idx_shop_price(shop, price)联合索引temu店群自动化报活动案例# 能用索引cursor.execute(WHERE shop 京东)# ✓ 用到shop列cursor.execute(WHERE shop 京东 AND price 100)# ✓ 用到shopprice# 不能用索引cursor.execute(WHERE price 100)# ✗ 跳过了shop索引失效cursor.execute(WHERE price 100 AND shop 京东)# ✓ SQLite优化器会自动调整顺序建联合索引时把选择性高的列不同值多的列放前面。比如shop有100个不同值status只有5个就把shop放前面。坑4索引不是越多越好每个索引都会拖慢写入速度INSERT/UPDATE要同步更新索引。如果一张表有10个索引写入速度可能下降50%。原则读多写少的表多建索引如日志表、统计表写多读少的表少建索引如实时采集表只对WHERE条件中频繁出现的列建索引定期清理无用索引# 查看索引使用情况MySQLcursor.execute( SELECT index_name, table_name, rows_read FROM sys.schema_index_statistics WHERE table_schema your_db ORDER BY rows_read ASC )# rows_read为0的索引可能从未被使用可以考虑删除坑5COUNT(*) 在大表上很慢SQLite的COUNT(*)是全表扫描100万行可能要几秒。如果你只需要知道有没有数据或大约多少条# 只判断是否有数据快cursor.execute(SELECT 1 FROM products LIMIT 1)has_datacursor.fetchone()isnotNone# 大致数量快但可能不精确cursor.execute(SELECT MAX(rowid) FROM products)approx_countcursor.fetchone()[0]# 精确数量慢cursor.execute(SELECT COUNT(*) FROM products)exact_countcursor.fetchone()[0]如果频繁需要总数维护一个计数器表每次增删时更新计数查询时直接读取。
