影刀RPA数据库操作实战：SQLite+MySQL企业级应用指南

影刀RPA数据库操作实战：SQLite+MySQL企业级应用指南

作者：林焱 | 影刀RPA教程系列第102篇

导读

前面很多文章都提到"数据存到Excel"。但当数据量达到万级以上，或者需要多流程共享数据、多用户并发访问时，Excel就不够用了——你需要数据库。

这篇文章从零开始，教你用影刀RPA操作数据库：

• SQLite（轻量级，无需安装服务器，单文件数据库）
• MySQL（企业级，适合多用户、大数据量场景）

包含完整的CRUD（增删改查）、事务处理、批量入库、查询优化等实战内容。

一、为什么需要数据库？

Excel vs 数据库对比

适用场景选择

数据量 < 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;![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/8053182d1d3f429fb49b31f5650c5120.png#pic_center)-- 条件更新（标记已处理的状态）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操作（企业级场景）

准备工作

1. 确保MySQL已安装并运行（公司IT部门一般都有）
2. 拿到连接信息：
   • 主机地址：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：待处理订单统计

-- 各种状态的订单数量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. 发送邮件通知：今日同步了多少文件、多少条数据

四、性能优化技巧

批量插入优化示例

【优化前：逐条INSERT（10000条 ≈ 600秒）】 对 data_table 循环： 执行SQL: INSERT INTO ... VALUES ... 循环结束 【优化后：批量INSERT（10000条 ≈ 10秒）】 拖入"批量插入数据表到数据库": 数据表 = data_table（一次性全部写入） 批次大小 = 1000（每批1000条，分10批提交）

五、常见问题与解决方案

Q1：SQLite数据库文件被锁住了？

原因：上一个流程没正确关闭连接，或者另一个进程正在读写这个文件。
解决：

1. 结束所有正在运行的RPA流程
2. 如果还不行：重启影刀RPA客户端
3. 终极方案：删除.db-wal和.db-shm文件（⚠️ 可能丢失未提交的数据）

Q2：MySQL连接超时怎么办？

原因：默认MySQL连接8小时空闲后会自动断开。
解决：

1. 每次操作前检查连接是否还活着（执行一个简单的SELECT 1）
2. 如果断开了，重新连接
3. 或者：在MySQL服务端设置wait_timeout = 31536000（一年不断开，仅开发环境）

Q3：中文乱码？

原因：编码不一致（Excel是GBK，数据库是UTF-8）。
解决：

1. 读取Excel时指定编码为UTF-8（如果Excel本身就是UTF-8）
2. 写入数据库前统一转UTF-8
3. MySQL连接时指定字符集：charset=utf8mb4

六、总结速查卡

如果这篇文章对你有帮助，欢迎分享给更多朋友！

下一篇，我们将讲解"影刀RPA调试进阶：复杂BUG排查方法论与工具使用技巧"。