影刀RPA数据库操作实战：SQLite+MySQL企业级应用指南
作者：林焱 | 影刀RPA教程系列第102篇
导读
前面很多文章都提到"数据存到Excel"。但当数据量达到万级以上，或者需要多流程共享数据、多用户并发访问时，Excel就不够用了——你需要数据库。
这篇文章从零开始，教你用影刀RPA操作数据库：
- SQLite（轻量级，无需安装服务器，单文件数据库）
- MySQL（企业级，适合多用户、大数据量场景）
包含完整的CRUD（增删改查）、事务处理、批量入库、查询优化等实战内容。
一、为什么需要数据库？
Excel vs 数据库对比
| 维度 | Excel | SQLite | MySQL |
|---|---|---|---|
| 数据量上限 | 约100万行 | 理论无限制（TB级） | PB级 |
| 并发访问 | 不支持（文件锁） | 有限支持 | 强支持 |
| 查询速度 | 万行以下快 | 百万行以内很快 | 亿行级别优化空间大 |
| 部署难度 | 无需部署 | 无需部署 | 需要装MySQL服务 |
| 多流程共享 | 困难（文件锁冲突） | 可以（连接池） | 天然支持 |
| 数据类型约束 | 弱（什么都能填） | 有基本约束 | 完整的约束系统 |
适用场景选择
数据量 < 1万条 → 用Excel就够了 数据量 1万~100万条 → 用SQLite（推荐！） 数据量 > 100万条 或 多人同时用 → 用MySQL二、Part A：SQLite操作（推荐入门首选）
什么是SQLite？
- 一个文件就是一个数据库（如
data.db） - 不需要安装任何服务
- 影刀RPA内置支持SQLite
- 适合个人使用、中小型项目
Step 1：创建/打开数据库
【指令】"打开/创建SQLite数据库" 参数： - 文件路径：C:\RPA_Data\my_database.db - 如果不存在则自动创建：✅ 勾选 → 输出变量：db_connection（数据库连接对象）💡 第一次运行时自动创建
my.db文件。后续所有操作都用同一个连接对象。
Step 2：建表
假设我们要建一个"商品价格监控"的数据表：
CREATETABLEIFNOTEXISTSproducts(idINTEGERPRIMARYKEYAUTOINCREMENT,-- 自增主键product_nameTEXTNOTNULL,-- 商品名称platformTEXTNOTNULL,-- 平台（京东/淘宝/拼多多）priceREALNOTNULL,-- 价格（浮点数）currencyTEXTDEFAULT'CNY',-- 货币单位[video(video-YDeurRa6-1782426071778)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限！)]urlTEXT,-- 商品链接collected_atDATETIMEDEFAULTCURRENT_TIMESTAMP,-- 采集时间UNIQUE(product_name,platform,collected_at)-- 同一商品同平台同一时间不重复);在影刀RPA中执行SQL：
拖入"执行SQL语句"指令：
- 连接对象：
db_connection - SQL语句：（粘贴上面的CREATE TABLE语句）
- 输出：执行结果（建表操作一般没有返回值）
Step 3：插入数据（Create）
方式1：插入单条
INSERTINTOproducts(product_name,platform,price,url)VALUES('iPhone 16 Pro','京东',8999.00,'https://item.jd.com/123456.html');方式2：批量插入（高性能！）
当你要一次插入1000条数据时，不要用循环逐条INSERT（太慢了），用批量插入：
-- 影刀RPA支持把数据表直接批量写入数据库-- 指令："批量插入数据表到数据库"参数： 连接对象=db_connection 表名="products"数据表=collected_data（之前采集到的数据表） 列映射："商品名称列"→ product_name"平台列"→ platform"价格列"→ price"链接列"→ url性能对比：
- 循环逐条INSERT 1000条 → 约60秒
- 批量插入 1000条 → 约2秒⚡️ （快30倍！）
Step 4：查询数据（Read）
基础查询：
-- 查询所有数据SELECT*FROMproducts;-- 查询最新10条SELECT*FROMproductsORDERBYcollected_atDESCLIMIT10;-- 按条件筛选SELECT*FROMproductsWHEREplatform='京东'ANDprice<5000;聚合查询（统计）：
-- 各平台平均价格SELECTplatform,AVG(price)asavg_price,COUNT(*)ascountFROMproductsGROUPBYplatform;-- 价格最低的前5名商品SELECTproduct_name,platform,MIN(price)aslowest_priceFROMproductsGROUPBYproduct_nameORDERBYlowest_priceASCLIMIT5;-- 每天采集的商品数量趋势SELECTDATE(collected_at)asdate,COUNT(*)asdaily_countFROMproductsGROUPBYDATE(collected_at)ORDERBYdateDESCLIMIT30;在影刀RPA中执行查询并获取结果：
拖入"执行查询SQL"指令：
- 连接对象：
db_connection - SQL语句：
SELECT * FROM products WHERE price < 5000 - 输出：
query_result（数据表，可以直接用于后续操作）
Step 5：更新数据（Update）
-- 更新某个商品的价格（采集到新价格后更新）UPDATEproductsSETprice=8500.00WHEREproduct_name='iPhone 16 Pro'ANDplatform='京东';-- 批量更新（给所有价格加10%模拟涨价）UPDATEproductsSETprice=price*1.1;-- 条件更新（标记已处理的状态）UPDATEproductsSETstatus='processed'WHEREstatus='pending';Step 6：删除数据（Delete）
⚠️注意：删除前一定要备份或加WHERE条件！否则会清空整张表！
-- 删除30天前的旧数据（定期清理）DELETEFROMproductsWHEREcollected_at<datetime('now','-30 days');-- 删除重复数据（保留id最小的那一条）DELETEFROMproductsWHEREidNOTIN(SELECTMIN(id)FROMproductsGROUPBYproduct_name,platform);-- 清空整张表（慎用！）-- DELETE FROM products; ← 这会删掉所有数据！Step 7：事务处理（保证数据一致性）
什么是事务？
事务就是"一组操作，要么全部成功，要么全部失败"，不会出现"只成功了一半"的情况。
场景举例：你要同时往两张表里写数据（一张是商品表，一张是日志表）。如果商品表写入成功了但日志表失败了，数据就"不一致"了。
事务的使用方法：
【开始事务】 拖入"开启事务"指令：connection = db_connection 【执行一系列SQL】 执行SQL: INSERT INTO products (...) VALUES (...) 执行SQL: INSERT INTO operation_log (action, time) VALUES ('insert_product', ...) 执行SQL: UPDATE statistics SET total_count = total_count + 1 【判断是否全部成功】 如果 所有SQL都执行成功： 【提交事务】 拖入"提交事务"指令：connection = db_connection 记录日志："事务提交成功" 否则： 【回滚事务】 拖入"回滚事务"指令：connection = db_connection 记录日志："事务回滚，所有修改已撤销" 【结束】三、Part B：MySQL操作（企业级场景）
准备工作
- 确保MySQL已安装并运行（公司IT部门一般都有）
- 拿到连接信息：
- 主机地址：
192.168.1.100（或localhost本机） - 端口：
3306 - 用户名：
rpa_user - 密码：
your_password - 数据库名：
rpa_database
- 主机地址：
Step 1：连接MySQL
拖入"连接MySQL数据库"指令：
- 主机：
192.168.1.100 - 端口：
3306 - 用户名：
rpa_user - 密码：
your_password - 数据库名：
rpa_database - 输出：
mysql_conn（连接对象）
Step 2：建表（MySQL语法略有不同）
CREATETABLEIFNOTEXISTSorders(idINTPRIMARYKEYAUTO_INCREMENT,order_noVARCHAR(50)NOTNULLUNIQUE,-- 订单号（唯一约束）customer_nameVARCHAR(100)NOTNULL,-- 客户姓名product_nameVARCHAR(200),-- 产品名称amountDECIMAL(12,2)NOTNULL,-- 金额（精确到分）order_statusENUM('pending','paid','shipped','completed','cancelled')DEFAULT'pending',create_timeDATETIMEDEFAULTCURRENT_TIMESTAMP,update_timeDATETIMEDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,INDEXidx_customer(customer_name),-- 索引（加速按客户名查询）INDEXidx_status(order_status),-- 索引（加速按状态查询）INDEXidx_create_time(create_time)-- 索引（加速按时间范围查询）);Step 3：批量导入Excel数据到MySQL
这是最常用的场景之一：把Excel里的订单数据批量导入MySQL。
【读取Excel】 读取Excel C:\Orders\orders_202606.xlsx → excel_data（数据表） 【数据清洗】 对 excel_data 每行循环： 如果 [当前行.金额] 为空 或 不是数字： [当前行.金额] = 0 如果 [当前行.订单号] 为空： 跳过此行（跳过无效数据） 循环结束 【批量写入MySQL】 拖入"批量插入数据表到MySQL"指令： 连接对象 = mysql_conn 表名 = "orders" 数据表 = excel_data 列映射： "订单号列" → order_no "客户名列" → customer_name "产品名列" → product_name "金额列" → amount 冲突策略：IGNORE（如果订单号已存在则跳过，不报错） 记录日志："成功导入" + 成功行数 + "条订单数据"Step 4：复杂查询与数据分析
场景1：月度销售报表
-- 按月份统计销售额和订单数SELECTDATE_FORMAT(create_time,'%Y-%m')asmonth,COUNT(*)asorder_count,SUM(amount)astotal_amount,AVG(amount)asavg_amount,MAX(amount)asmax_amountFROMordersWHEREcreate_time>='2026-01-01'GROUPBYDATE_FORMAT(create_time,'%Y-%m')ORDERBYmonthDESC;场景2：TOP10客户排行
-- 消费金额最高的前10个客户SELECTcustomer_name,COUNT(*)asorder_count,SUM(amount)astotal_spent,AVG(amount)asavg_order_amountFROMordersWHEREorder_status!='cancelled'GROUPBYcustomer_nameORDERBYtotal_spentDESCLIMIT10;场景3：待处理订单统计
temu店群自动化报活动案例
-- 各种状态的订单数量SELECTorder_status,COUNT(*)ascount,SUM(amount)astotal_amountFROMordersGROUPBYorder_statusORDERBYcountDESC;Step 5：定时同步任务（Excel → MySQL）
很多公司的业务数据还在Excel里流转，需要每天定时把最新的Excel数据同步到MySQL数据库，供BI系统分析。
【主流程设计】 1. 定义配置： excel_dir = "C:\Orders\" -- Excel文件存放目录 mysql_conn = 连接MySQL -- 已建立的连接 sync_log_file = "C:\Logs\sync.log" 2. 获取目录下所有Excel文件列表 3. 对每个Excel文件循环： a. 读取Excel → data_table b. 文件名解析出日期（如 orders_20260610.xlsx → 2026-06-10） c. 在data_table中新增"import_date"列，填入解析出的日期 d. 批量写入MySQL（冲突策略=UPSERT，即存在就更新，不存在就插入） e. 移动已处理的Excel文件到"已完成"文件夹 f. 记录同步日志 4. 关闭MySQL连接 5. 发送邮件通知：今日同步了多少文件、多少条数据四、性能优化技巧
| 场景 | 优化方案 | 效果提升 |
|---|---|---|
| 大批量插入（万级） | 先关闭索引 → 批量插入 → 重建索引 | 5~10倍 |
| 查询慢 | 给常用查询字段加INDEX | 10~100倍 |
| 全文搜索慢 | 用LIKE ‘%关键词%’ 很慢 | 改用FTS全文索引 |
| 数据库文件太大 | 定期归档旧数据（移到历史表） | 保持活跃数据量小 |
| 连接频繁建立销毁 | 使用连接池（复用连接） | 减少开销 |
批量插入优化示例
【优化前：逐条INSERT（10000条 ≈ 600秒）】 对 data_table 循环： 执行SQL: INSERT INTO ... VALUES ... 循环结束 【优化后：批量INSERT（10000条 ≈ 10秒）】 拖入"批量插入数据表到数据库": 数据表 = data_table（一次性全部写入） 批次大小 = 1000（每批1000条，分10批提交）五、常见问题与解决方案
Q1：SQLite数据库文件被锁住了？
原因：上一个流程没正确关闭连接，或者另一个进程正在读写这个文件。
解决：
- 结束所有正在运行的RPA流程
- 如果还不行：重启影刀RPA客户端
- 终极方案：删除
.db-wal和.db-shm文件（⚠️ 可能丢失未提交的数据）
Q2：MySQL连接超时怎么办？
原因：默认MySQL连接8小时空闲后会自动断开。
解决：
- 每次操作前检查连接是否还活着（执行一个简单的
SELECT 1） - 如果断开了，重新连接
- 或者：在MySQL服务端设置
wait_timeout = 31536000（一年不断开，仅开发环境）
Q3：中文乱码？
原因：编码不一致（Excel是GBK，数据库是UTF-8）。
解决：
- 读取Excel时指定编码为UTF-8（如果Excel本身就是UTF-8）
- 写入数据库前统一转UTF-8
- MySQL连接时指定字符集：
charset=utf8mb4
六、总结速查卡
| 操作 | SQLite指令 | MySQL指令 |
|---|---|---|
| 创建/打开数据库 | 打开SQLite数据库 | 连接MySQL数据库 |
| 建表 | CREATE TABLE | CREATE TABLE（语法略不同） |
| 单条插入 | INSERT INTO | INSERT INTO |
| 批量插入 | 批量插入数据表 | 批量插入数据表到MySQL |
| 查询 | SELECT * FROM | SELECT * FROM |
| 更新 | UPDATE … SET | UPDATE … SET |
| 删除 | DELETE FROM | DELETE FROM |
| 事务 | 开启/提交/回滚事务 | 同左 |
| 关闭连接 | 关闭SQLite连接 | 关闭MySQL连接 |
如果这篇文章对你有帮助，欢迎分享给更多朋友！
下一篇，我们将讲解"影刀RPA调试进阶：复杂BUG排查方法论与工具使用技巧"。
