SQLite适合单机、小规模数据存储，但一旦你的RPA项目需要多人共享数据、远程访问、高并发写入，SQLite就不够用了。这时候需要切换到MySQL——真正的“客户端-服务器”数据库，支持多用户同时读写、支持远程连接、支持千万级数据量。
我之前帮一个团队做“多店铺销售数据汇总”项目，一开始用的SQLite，数据文件存在共享文件夹里。两个采集流程同时跑的时候频繁报错“数据库被锁定”，后来换成MySQL，问题直接消失，写入速度还快了两倍。
这篇文章不讲MySQL的理论知识，就讲影刀RPA里怎么连接MySQL、怎么读写数据、怎么用Pandas和MySQL互通。读完你可以把“本地SQLite”升级成“云端MySQL”，让数据在不同流程和不同机器之间共享。
什么时候该用MySQL
| 场景 | 用SQLite | 用MySQL |
|---|---|---|
| 单机单流程 | ✅ 合适 | 杀鸡用牛刀 |
| 多流程同时读写 | ❌ 会锁库 | ✅ 支持并发 |
| 多台电脑共享数据 | ❌ 文件共享不稳定 | ✅ 服务器集中存储 |
| 数据量 > 100万行 | ⚠️ 开始变慢 | ✅ 流畅 |
| 需要远程访问 | ❌ 不支持 | ✅ 原生支持 |
| 需要用户权限管理 | ❌ 弱 | ✅ 完善的权限体系 |
核心判断标准：如果你的数据需要在“多台电脑”或“多个流程”之间共享，就应该用MySQL。
第一步：安装MySQL和Python驱动
安装MySQL数据库（服务器端）
如果公司已有MySQL服务器，直接用。如果没有，在电脑上安装一个：
- 去MySQL官网下载MySQL Community Server（社区版，免费）
- 安装时设置root密码（务必记好）
- 安装完成后，用命令行或Navicat测试连接
云数据库方案（推荐，省心）：
如果不想自己维护MySQL服务器，用云服务：
- 阿里云RDS MySQL
- 腾讯云MySQL
- 华为云GaussDB
按月付费，不用管运维，适合项目型场景。
在影刀里安装Python驱动
影刀的Python环境默认没有安装MySQL驱动，需要手动安装。
在“执行Python代码”指令里运行：
import subprocess
import sys
# 安装pymysql（轻量级MySQL驱动）
subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pymysql', '-i', 'https://pypi.tuna.tsinghua.edu.cn/simple'])
print("✅ pymysql 安装完成")
# 如果要用Pandas读写MySQL，还需要安装
subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pandas', 'sqlalchemy', '-i', 'https://pypi.tuna.tsinghua.edu.cn/simple'])
print("✅ pandas + sqlalchemy 安装完成")
注意： 第一次运行时，需要在影刀顶部的Python图标点亮状态下执行这段代码。安装成功后，后续流程就可以直接import pymysql了。
第二步：创建数据库和表（首次运行）
# ============================================
# 执行Python代码：创建MySQL数据库和商品表
# 输入：
# - host（文本）：数据库地址，如 "localhost" 或 "192.168.1.100"
# - user（文本）：用户名，如 "root"
# - password（文本）：密码
# - database（文本）：数据库名，如 "rpa_data"
# 输出：无
# ============================================
import pymysql
def init_mysql(host, user, password, database):
# 先连接到MySQL（不指定数据库）
conn = pymysql.connect(
host=host,
user=user,
password=password,
charset='utf8mb4'
)
cursor = conn.cursor()
# 创建数据库（如果不存在）
cursor.execute(f"CREATE DATABASE IF NOT EXISTS {database} DEFAULT CHARSET utf8mb4")
print(f"✅ 数据库 {database} 已创建")
# 切换到目标数据库
conn.select_db(database)
# 创建商品数据表（如果不存在）
create_table_sql = """
CREATE TABLE IF NOT EXISTS products (
id INT AUTO_INCREMENT PRIMARY KEY,
platform VARCHAR(20) NOT NULL COMMENT '平台名称',
title VARCHAR(500) NOT NULL COMMENT '商品标题',
price DECIMAL(10,2) DEFAULT 0.00 COMMENT '价格',
sales INT DEFAULT 0 COMMENT '销量',
shop_name VARCHAR(200) COMMENT '店铺名称',
product_url VARCHAR(500) COMMENT '商品链接',
collect_date DATE COMMENT '采集日期',
collect_time TIME COMMENT '采集时间',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY unique_product (platform, product_url(255))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
"""
cursor.execute(create_table_sql)
print("✅ 数据表 products 已创建")
# 创建索引（提高查询速度）
cursor.execute("CREATE INDEX idx_date ON products(collect_date)")
cursor.execute("CREATE INDEX idx_platform ON products(platform)")
cursor.execute("CREATE INDEX idx_price ON products(price)")
print("✅ 索引创建完成")
conn.commit()
cursor.close()
conn.close()
print("✅ MySQL初始化完成")
# 调用
init_mysql(
host="localhost",
user="root",
password="your_password",
database="rpa_data"
)
表结构说明：
| 字段 | 类型 | 说明 |
|---|---|---|
| id | INT | 自增主键 |
| platform | VARCHAR(20) | 平台名称（拼多多/淘宝/小红书） |
| title | VARCHAR(500) | 商品标题（最长500字符） |
| price | DECIMAL(10,2) | 价格（10位数字，2位小数） |
| sales | INT | 销量（整数） |
| shop_name | VARCHAR(200) | 店铺名称 |
| product_url | VARCHAR(500) | 商品链接 |
| collect_date | DATE | 采集日期 |
| collect_time | TIME | 采集时间 |
| created_at | TIMESTAMP | 记录创建时间（自动） |
| updated_at | TIMESTAMP | 记录更新时间（自动） |
| UNIQUE key | (platform, product_url) | 同一平台同一商品不重复 |
和SQLite表的区别：
- MySQL支持
TIMESTAMP自动记录时间 - MySQL支持更精细的
DECIMAL浮点数精度 - MySQL的
UNIQUE KEY写法略有不同 product_url在UNIQUE里加了长度限制（255字符），因为索引字段不能太长
第三步：批量写入数据到MySQL
# ============================================
# 执行Python代码：批量插入数据到MySQL
# 输入：
# - data_list（二维列表）：每条数据格式：
# [platform, title, price, sales, shop_name, product_url, collect_date, collect_time]
# - mysql_config（字典）：{host, user, password, database}
# 输出：insert_count（数字）
# ============================================
import pymysql
def batch_insert_mysql(data_list, mysql_config):
if not data_list:
return 0
conn = pymysql.connect(
host=mysql_config['host'],
user=mysql_config['user'],
password=mysql_config['password'],
database=mysql_config['database'],
charset='utf8mb4'
)
cursor = conn.cursor()
# 插入或忽略重复
insert_sql = """
INSERT IGNORE INTO products
(platform, title, price, sales, shop_name, product_url, collect_date, collect_time)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
"""
success_count = 0
fail_count = 0
for row in data_list:
try:
cursor.execute(insert_sql, (
row[0], # platform
row[1], # title
row[2], # price
row[3], # sales
row[4], # shop_name
row[5], # product_url
row[6], # collect_date
row[7] # collect_time
))
success_count += 1
except Exception as e:
fail_count += 1
print(f"插入失败：{row[1]}，错误：{e}")
conn.commit()
cursor.close()
conn.close()
print(f"✅ 批量插入完成：成功{success_count}条，失败{fail_count}条")
return success_count
关键写法：
INSERT IGNORE INTO 表示如果触发了UNIQUE KEY约束（同一平台同一商品已存在），这条数据会被忽略，不报错。这是实现“增量写入、不重复”的核心。
第四步：从MySQL查询数据到影刀
# ============================================
# 执行Python代码：从MySQL查询数据
# 输入：
# - query_date（文本）：查询日期，格式 YYYY-MM-DD
# - mysql_config（字典）：数据库连接配置
# 输出：result_data（二维列表），第一行是表头
# ============================================
import pymysql
def query_mysql(query_date, mysql_config):
conn = pymysql.connect(
host=mysql_config['host'],
user=mysql_config['user'],
password=mysql_config['password'],
database=mysql_config['database'],
charset='utf8mb4'
)
cursor = conn.cursor()
# 查询指定日期的数据，按价格从高到低排序
sql = """
SELECT platform, title, price, sales, shop_name, collect_time
FROM products
WHERE collect_date = %s
ORDER BY price DESC
"""
cursor.execute(sql, (query_date,))
# 获取所有行
rows = cursor.fetchall()
# 加上表头
columns = ["平台", "标题", "价格", "销量", "店铺", "采集时间"]
result = [columns] + [list(row) for row in rows]
cursor.close()
conn.close()
print(f"✅ 查询到 {len(rows)} 条记录")
return result
# 调用
result_data = query_mysql("2026-07-06", mysql_config)
在影刀里的用法： 查询结果可以直接用“写入表格数据”写入Excel，或者作为流程下一步的数据源。
第五步：Pandas + MySQL 数据互通（高级）
Pandas可以直接读写MySQL，省去手动拼接SQL的麻烦。
# ============================================
# 执行Python代码：Pandas读写MySQL
# 输入：
# - mysql_config（字典）
# - collect_date（文本）
# 输出：
# - df_json（JSON字符串），Pandas DataFrame的JSON格式
# ============================================
import pandas as pd
from sqlalchemy import create_engine
import json
def pandas_mysql_demo(mysql_config, collect_date):
# 创建数据库连接（SQLAlchemy）
db_url = f"mysql+pymysql://{mysql_config['user']}:{mysql_config['password']}@{mysql_config['host']}/{mysql_config['database']}?charset=utf8mb4"
engine = create_engine(db_url)
# ===== 读取：MySQL → Pandas =====
df = pd.read_sql(
f"SELECT * FROM products WHERE collect_date = '{collect_date}' ORDER BY price DESC",
engine
)
print(f"✅ 从MySQL读取 {len(df)} 条数据到Pandas")
# ===== 用Pandas做数据分析 =====
# 各平台平均价格
avg_price_by_platform = df.groupby('platform')['price'].mean()
print("各平台平均价格：")
print(avg_price_by_platform)
# 销量TOP10
top10 = df.nlargest(10, 'sales')[['title', 'sales', 'price']]
print("销量TOP10：")
print(top10)
# ===== 写入：Pandas → MySQL（新表） =====
# 把分析结果写入一个新表
top10.to_sql('top10_products', engine, if_exists='replace', index=False)
print("✅ TOP10数据已写入MySQL表 top10_products")
# ===== 返回DataFrame的JSON（给影刀用） =====
return df.to_json(orient='records', force_ascii=False)
# 调用
df_json = pandas_mysql_demo(mysql_config, "2026-07-06")
这个写法的价值：
- 用Pandas做数据透视、分组、排序、去重，一行代码顶影刀十行循环
- 分析结果可以直接写回MySQL，也可以传给影刀做后续处理
- 适合“采集→分析→存储”一体化流程
第六步：MySQL连接池配置（多流程共享）
如果多个流程同时访问MySQL，每次都新建连接、断开连接，效率很低。
推荐：在MySQL连接配置中开启连接池。
# 用SQLAlchemy的连接池
from sqlalchemy import create_engine
# 连接池配置
engine = create_engine(
db_url,
pool_size=5, # 连接池大小
max_overflow=10, # 最大额外连接数
pool_recycle=3600, # 连接回收时间（秒）
pool_pre_ping=True # 使用前检查连接是否有效
)
# 所有读写操作都用这个engine，不要反复创建
在影刀里的实践：
- 把
engine创建在“执行Python代码”的外部（比如做成全局变量），不要让每次调用都重新创建连接 - 或者在主流程里创建一次连接，传递给多个子流程使用
MySQL常用SQL速查表
| 操作 | SQL语句 | 说明 |
|---|---|---|
| 插入（忽略重复） | INSERT IGNORE INTO 表名 VALUES (...) | 已存在则跳过 |
| 插入（更新重复） | INSERT ... ON DUPLICATE KEY UPDATE 价格=VALUES(价格) | 已存在则更新价格 |
| 查询全部 | SELECT * FROM 表名 | 返回所有数据 |
| 条件查询 | SELECT * FROM 表名 WHERE 价格 > 50 | 按条件筛选 |
| 排序查询 | SELECT * FROM 表名 ORDER BY 销量 DESC LIMIT 20 | 销量TOP20 |
| 分组统计 | SELECT 平台, COUNT(*) FROM 表名 GROUP BY 平台 | 按平台统计数量 |
| 日期范围 | SELECT * FROM 表名 WHERE 采集日期 BETWEEN '2026-07-01' AND '2026-07-07' | 按日期区间筛选 |
| 模糊匹配 | SELECT * FROM 表名 WHERE 标题 LIKE '%耳机%' | 标题包含“耳机” |
| 删除旧数据 | DELETE FROM 表名 WHERE 采集日期 < '2026-06-01' | 删除指定日期前数据 |
| 清空表 | TRUNCATE TABLE 表名 | 删除所有数据（比DELETE快） |
容易踩的坑速查表
| 现象 | 排查方向 | 修复方法 |
|---|---|---|
| 连接报错“Access denied” | 用户名或密码错了 | 检查user和password是否正确 |
| 连接报错“Can't connect” | MySQL服务没启动或地址错了 | 检查host是否正确，MySQL是否运行中 |
| 连接报错“Unknown database” | 数据库名写错了 | 先用SQLite风格创建数据库，或检查database名 |
| 插入中文报错乱码 | 字符集没设好 | 连接时指定charset='utf8mb4' |
| 插入报错“Data too long” | 字段长度不够 | 检查表结构，加长VARCHAR长度 |
| 查询返回空列表 | 日期格式不匹配 | 确保传入的日期和数据库里存的格式一致（YYYY-MM-DD） |
| 多流程同时写入卡死 | 连接池满了 | 增大pool_size或减少并发数 |
| 数据在MySQL里能看到，但影刀读不到 | SQL语句的条件写错了 | 在Navicat里验证SQL语句是否正确 |
MySQL vs SQLite 选择指南
| 场景 | 用SQLite | 用MySQL |
|---|---|---|
| 单个流程、本地数据 | ✅ | ❌ |
| 多个流程、同一台电脑 | ⚠️ 可能锁库 | ✅ |
| 多台电脑、数据共享 | ❌ 无法共享 | ✅ 远程访问 |
| 数据量 < 10万行 | ✅ | ✅ |
| 数据量 > 100万行 | ⚠️ | ✅ |
| 需要数据备份/恢复 | ⚠️ 手动备份 | ✅ 原生支持 |
| 需要图形化管理工具 | SQLiteStudio | Navicat/DBeaver |
| 部署和运维成本 | 零（一个文件） | 需要安装配置 |
一条核心建议
迁移到MySQL的时机：当你开始出现“数据库被锁定”报错，或者需要在多台电脑之间共享数据时，就该迁移了。
迁移的路径建议：
- 第一步：先搭一个本地MySQL，把现有SQLite数据导入进去（用
INSERT ... SELECT或导出CSV再导入） - 第二步：把流程里的数据库连接地址从
products.db改成localhost:3306 - 第三步：确认流程能正常读写MySQL
- 第四步：把MySQL迁移到云服务器，让所有电脑共享
迁移过程中最容易踩的坑是“字符集”——中文显示为乱码。解决办法：所有地方统一用utf8mb4（MySQL的UTF-8完整版），连接时指定charset='utf8mb4'，表的字符集也设成utf8mb4。
另外，永远不要在流程里硬编码数据库密码。用流程参数或环境变量存储敏感信息，避免密码泄露。
推荐资源
- MySQL官方文档：安装指南
- Navicat / DBeaver（MySQL图形化管理工具，比命令行直观）
- 影刀社区示例：“MySQL数据存储模板”（搜索“MySQL”）
#影刀RPA #RPA自动化 #MySQL #数据库 #Pandas
作者：林焱
本文为《影刀RPA学习手册》系列文章之一，内容源于实操经验的整理与分享。