Python批量导入CSV到MySQL的完整步骤
作者：小小的木头人
本文分享一个Python自动导入方案,能自动扫描目录、根据文件名匹配表名、自动检测文件编码（GBK/UTF-8等）,并利用pandas分批读取避免内存溢出,同时显示进度条让你实时掌握导入进度,适合运维、数据迁移等场景,帮你高效完成批量数据导入,需要的朋友可以参考下
下面给你整理成一篇 适合技术博客发布的版本（结构清晰、运维/数据工程风格）。内容包括背景、思路、代码、优化点等。
Python 批量导入 CSV 到 MySQL（自动识别编码 + 进度条）
在实际工作中，经常会遇到这样的需求：
- 已知 MySQL 连接信息
- 目录中有 大量 CSV 文件
- 文件名规则类似：
deps_01.csv deps_02.csv deps_03.csv
需要把这些 CSV 批量导入到 MySQL 指定表中。
并且存在几个实际问题：
- CSV 文件数量多
- 文件可能比较大
- 编码可能不统一（GBK / GB2312 / UTF-8 等）
- 希望看到 导入进度
本文提供一套 Python 自动导入方案。
实现功能：
- 自动扫描 CSV 目录
- 根据 文件名匹配表名
- 自动检测 CSV 编码
- 分批导入（避免内存爆）
- 显示导入进度条
一、环境准备
安装 Python 依赖：
pip install pandas sqlalchemy pymysql tqdm chardet
依赖说明：
| 库 | 作用 |
|---|---|
| pandas | 读取 CSV |
| sqlalchemy | 数据库连接 |
| pymysql | MySQL 驱动 |
| tqdm | 进度条 |
| chardet | 检测文件编码 |
二、目录结构示例
例如 CSV 文件目录：
D:\test deps_01.csv deps_02.csv deps_03.csv user_01.csv user_02.csv
如果配置：
TABLE_LIST = ["user"]
则：
user_01.csv user_02.csv
会导入到 user 表。
三、完整 Python 脚本
import os
import pandas as pd
from sqlalchemy import create_engine
from tqdm import tqdm
import chardet
# ===============================
# MySQL 配置
# ===============================
MYSQL_USER = "root"
MYSQL_PASS = "123456"
MYSQL_HOST = "127.0.0.1"
MYSQL_PORT = 3306
MYSQL_DB = "testdb"
# ===============================
# CSV 目录
# ===============================
CSV_DIR = r"D:\test"
# ===============================
# 需要导入的表
# ===============================
TABLE_LIST = [
"user"
]
# ===============================
# 每批导入行数
# ===============================
CHUNK_SIZE = 5000
# ===============================
# 创建数据库连接
# ===============================
engine = create_engine(
f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASS}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}?charset=utf8mb4"
)
# ===============================
# 自动识别编码
# ===============================
def detect_encoding(file_path):
with open(file_path, "rb") as f:
data = f.read(100000) # 只读取前100KB
result = chardet.detect(data)
return result["encoding"]
print("程序启动，扫描目录:", CSV_DIR)
# ===============================
# 扫描CSV文件
# ===============================
for file in os.listdir(CSV_DIR):
if not file.endswith(".csv"):
continue
# 根据文件名匹配表名
matched_table = None
for table in TABLE_LIST:
if table in file:
matched_table = table
break
if not matched_table:
continue
filepath = os.path.join(CSV_DIR, file)
print(f"\n准备导入 {file} -> 表 {matched_table}")
try:
encoding = detect_encoding(filepath)
print("检测到编码:", encoding)
reader = pd.read_csv(
filepath,
encoding=encoding,
chunksize=CHUNK_SIZE,
low_memory=False
)
for chunk in tqdm(reader, desc=file):
chunk.to_sql(
name=matched_table,
con=engine,
if_exists="append",
index=False
)
print(f"{file} 导入完成")
except Exception as e:
print(f"{file} 导入失败:", e)
print("\n所有任务完成")
四、运行脚本
执行：
python import_csv_to_mysql.py
运行效果：
程序启动，扫描目录: D:\test 准备导入 user_01.csv -> 表 user 检测到编码: GB18030 user_01.csv: 120it [00:20, 5.8it/s] user_01.csv 导入完成
其中：
120it
表示 处理了 120 个批次。
如果：
CHUNK_SIZE = 5000
则：
120 × 5000 ≈ 60万行数据
五、关键实现思路
1 自动检测 CSV 编码
很多 CSV 来自：
- Excel
- Oracle 导出
- Windows 程序
编码可能是：
UTF8 GBK GB2312 GB18030
使用：
chardet.detect()
自动识别。
2 分批导入避免内存溢出
使用：
chunksize = 5000
pandas 会 每次读取 5000 行。
好处：
- 内存稳定
- 支持超大 CSV
- 导入更安全
3 进度条显示
使用：
from tqdm import tqdm
示例：
deps_01.csv: 62it [01:46, 1.56s/it]
含义：
| 字段 | 含义 |
|---|---|
| 62it | 处理 62 个批次 |
| 1.56s/it | 每个批次约 1.56 秒 |
4 文件名自动匹配表
代码逻辑：
文件名包含表名 → 导入该表
例如：
| 文件名 | 导入表 |
|---|---|
| user_01.csv | user |
| user_02.csv | user |
六、性能优化建议
如果数据量很大（千万级），建议：
1 增大批量
CHUNK_SIZE = 10000
或
CHUNK_SIZE = 20000
2 关闭 MySQL 索引
导入前：
ALTER TABLE user DISABLE KEYS;
导入后：
ALTER TABLE user ENABLE KEYS;
3 使用 LOAD DATA（最快）
如果 CSV 非常大，可以使用：
LOAD DATA INFILE
性能比 pandas 快 10~50倍。
七、适用场景
该脚本适用于：
- 运维导入业务数据
- Oracle / Excel 导出数据迁移
- 批量 CSV 导入 MySQL
- 数据初始化
八、总结
本文实现了一套 自动化 CSV 导入 MySQL 工具，特点：
- 自动扫描目录
- 自动识别编码
- 按表名匹配
- 分批导入
- 进度条显示
适合处理 大量 CSV 文件导入数据库 的场景。
到此这篇关于Python批量导入CSV到MySQL的完整步骤的文章就介绍到这了,更多相关Python批量导入CSV到MySQL内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！