我要评论python将mysql数据库中所有表的数据都导出为csv文件到一个目录，并压缩为zip文件到另一个目录下，然后解压缩这个目录中的所有zip文件到第三个目录下。不使用pandas库，需要考虑sql结果集是大数据量分批数据导出的情况，通过多线程和异步操作来提高程序性能，程序需要异常处理和输出，输出出错时的错误信息，每次每个查询导出数据的运行状态和表数据行数以及运行时间戳，导出时间，输出每个文件记录数量的日志。
该脚本已在考虑大数据量、异常处理和性能优化的基础上进行了全面设计，能够处理大多数常见场景。根据具体需求可进一步调整批量大小（batch_size）和线程数（max_workers）以获得最佳性能。
import os
import csv
import zipfile
import logging
import mysql.connector
from datetime import datetime
import time
import concurrent.futures
import glob
# 配置日志
logging.basicconfig(
level=logging.info,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.filehandler('data_export.log'),
logging.streamhandler()
]
)
logger = logging.getlogger(__name__)
def export_table_to_csv(table_name, csv_path, db_config, batch_size=1000):
"""导出单个表的数据到csv文件，分批处理"""
conn = none
cursor = none
total_rows = 0
try:
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
# 获取数据并写入csv
with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
# 执行查询并获取列名
cursor.execute(f"select * from `{table_name}`")
columns = [col[0] for col in cursor.description]
writer.writerow(columns)
# 分批获取数据
while true:
rows = cursor.fetchmany(batch_size)
if not rows:
break
writer.writerows(rows)
total_rows += len(rows)
logger.debug(f"{table_name} 已导出 {total_rows} 行")
logger.info(f"{table_name} csv导出完成，总行数：{total_rows}")
return total_rows
except exception as e:
logger.error(f"导出表 {table_name} 失败: {str(e)}", exc_info=true)
raise
finally:
if cursor:
cursor.close()
if conn and conn.is_connected():
conn.close()
def compress_to_zip(source_path, zip_path):
"""压缩文件为zip格式"""
try:
with zipfile.zipfile(zip_path, 'w', zipfile.zip_deflated) as zipf:
zipf.write(source_path, arcname=os.path.basename(source_path))
logger.info(f"成功压缩 {source_path} 到 {zip_path}")
except exception as e:
logger.error(f"压缩 {source_path} 失败: {str(e)}", exc_info=true)
raise
def process_table(table_name, db_config, csv_dir, zip_dir):
"""处理单个表的导出和压缩"""
start_time = time.time()
logger.info(f"开始处理表: {table_name}")
status = "成功"
rows_exported = 0
try:
# 定义文件路径
csv_filename = f"{table_name}.csv"
zip_filename = f"{table_name}.zip"
csv_path = os.path.join(csv_dir, csv_filename)
zip_path = os.path.join(zip_dir, zip_filename)
# 导出csv
rows_exported = export_table_to_csv(table_name, csv_path, db_config)
# 压缩文件
compress_to_zip(csv_path, zip_path)
except exception as e:
status = f"失败: {str(e)}"
# 清理可能存在的中间文件
for path in [csv_path, zip_path]:
if path and os.path.exists(path):
try:
os.remove(path)
logger.warning(f"已清理文件: {path}")
except exception as clean_error:
logger.error(f"清理文件失败: {clean_error}")
finally:
duration = time.time() - start_time
log_message = (
f"表处理完成 - 表名: {table_name}, "
f"状态: {status}, "
f"导出行数: {rows_exported}, "
f"耗时: {duration:.2f}秒"
)
logger.info(log_message)
def unzip_files(zip_dir, unzip_dir):
"""解压指定目录中的所有zip文件"""
zip_files = glob.glob(os.path.join(zip_dir, '*.zip'))
if not zip_files:
logger.warning("未找到zip文件，跳过解压")
return
with concurrent.futures.threadpoolexecutor() as executor:
futures = []
for zip_path in zip_files:
futures.append(executor.submit(
lambda: extract_zip(zip_path, unzip_dir)
))
for future in concurrent.futures.as_completed(futures):
try:
future.result()
except exception as e:
logger.error(f"解压过程中发生错误: {str(e)}")
def extract_zip(zip_path, unzip_dir):
"""解压单个zip文件"""
try:
start_time = time.time()
with zipfile.zipfile(zip_path, 'r') as zip_ref:
zip_ref.extractall(unzip_dir)
duration = time.time() - start_time
logger.info(f"解压完成: {zip_path} => {unzip_dir} (耗时: {duration:.2f}秒)")
except exception as e:
logger.error(f"解压 {zip_path} 失败: {str(e)}", exc_info=true)
raise
def main():
# 配置参数
db_config = {
'host': 'localhost',
'user': 'your_username',
'password': 'your_password',
'database': 'your_database'
}
# 目录配置
base_dir = os.path.dirname(os.path.abspath(__file__))
csv_dir = os.path.join(base_dir, 'csv_exports')
zip_dir = os.path.join(base_dir, 'zip_archives')
unzip_dir = os.path.join(base_dir, 'unzipped_files')
# 创建目录
for dir_path in [csv_dir, zip_dir, unzip_dir]:
os.makedirs(dir_path, exist_ok=true)
logger.info(f"目录已准备: {dir_path}")
# 获取所有表名
try:
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor()
cursor.execute("show tables")
tables = [table[0] for table in cursor.fetchall()]
logger.info(f"发现 {len(tables)} 个需要处理的表")
except exception as e:
logger.error(f"获取数据库表失败: {str(e)}", exc_info=true)
return
finally:
if 'cursor' in locals():
cursor.close()
if 'conn' in locals() and conn.is_connected():
conn.close()
# 处理所有表（多线程导出和压缩）
with concurrent.futures.threadpoolexecutor(max_workers=4) as executor:
futures = []
for table in tables:
futures.append(executor.submit(
process_table,
table,
db_config,
csv_dir,
zip_dir
))
# 处理任务结果
for future in concurrent.futures.as_completed(futures):
try:
future.result()
except exception as e:
logger.error(f"表处理异常: {str(e)}")
# 解压所有zip文件（多线程解压）
logger.info("开始解压所有zip文件")
unzip_files(zip_dir, unzip_dir)
logger.info("全部处理流程完成")
if __name__ == "__main__":
main()
关键特性说明：
1.分批处理大数据：
- 使用fetchmany(batch_size)分批获取数据（默认每批1000行）
- 流式处理减少内存占用
2.多线程处理：
- 使用threadpoolexecutor并行处理不同表的导出和压缩
- 独立的数据库连接池（每个线程有自己的连接）
- 并行解压处理
3.异常处理：
- 全面的try-except块覆盖所有关键操作
- 自动清理失败时产生的中间文件
- 详细的错误日志记录（包含堆栈跟踪）
4.日志记录：
- 同时输出到文件和终端
- 记录时间戳、操作类型、状态、耗时等关键信息
- 包含每个表的处理结果统计
5.文件管理：
- 自动创建所需目录
- 使用zip_deflated进行高效压缩
- 安全的文件路径处理
6.性能优化：
- 使用服务器端游标避免内存过载
- 可配置的批量大小和线程数
- 异步i/o操作
使用说明：
安装依赖：
pip install mysql-connector-python
修改配置：
更新db_config中的数据库连接信息
根据需要调整目录路径（csv_dir, zip_dir, unzip_dir）
运行脚本：
python script.py
查看日志：
实时终端输出
详细日志文件data_export.log
扩展建议：
通过命令行参数接受数据库配置和目录路径
添加邮件通知功能（处理完成或失败时通知）
实现断点续传功能
添加文件校验（md5校验和）
支持配置文件（yaml/json格式）
添加进度条显示
到此这篇关于python实现将mysql中所有表的数据都导出为csv文件并压缩的文章就介绍到这了,更多相关python mysql数据导出为csv内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论