1 准备工作1.1 环境准备
操作系统：Microsoft Windows 10 专业工作站版
软件版本：Python 3.9.6
第三方包：
pip install pandas2.1.0
pip install pymysql1.1.0
pip install sqlalchemy==2.0.23
Get-WmiObject -Class Win32_OperatingSystem | Select-Object -Property Captionpython –versionpip install pandas==2.1.0 -i https://mirrors.aliyun.com/pypi/simple/pip install pymysql==1.1.0 -i https://mirrors.aliyun.com/pypi/simple/pip install sqlalchemy==2.0.23 -i https://mirrors.aliyun.com/pypi/simple/
1.2 数据库准备
数据库版本：Ver 8.1.0 for Win64 on x86_64 (MySQL Community Server – GPL)
mysql --versionmysql -h 127.0.0.1 -P 3306 -u root -p"123456"
1.3 数据准备
运行以下python代码准备模拟数据用于测试。
import osif not os.path.exists('./datas'): os.mkdir('./datas')text = '''+-------------+--------------+--------------+-------------+| customer_id | name | visited_on | amount |+-------------+--------------+--------------+-------------+| 1 | Jhon | 2019-01-01 | 100 || 2 | Daniel | 2019-01-02 | 110 || 3 | Jade | 2019-01-03 | 120 || 4 | Khaled | 2019-01-04 | 130 || 5 | Winston | 2019-01-05 | 110 | | 6 | Elvis | 2019-01-06 | 140 | | 7 | Anna | 2019-01-07 | 150 || 8 | Maria | 2019-01-08 | 80 || 9 | Jaze | 2019-01-09 | 110 | | 1 | Jhon | 2019-01-10 | 130 | | 3 | Jade | 2019-01-10 | 150 | +-------------+--------------+--------------+-------------+'''with open('./datas/customer.txt','w',encoding='utf-8') as file:file.write(text)运行前：
代码运行前如图，只有一个datas.py的文件。
代码运行后如图，在运行前的基础上生成了一个datas的文件夹以及一个存有用来测试的模拟数据文件，也可直接从mysql端复制粘贴并手动创建即可，值得注意的是不能有任何空行。
2 mysql端配置2.1 连接mysql
在powershell终端使用命令连接mysql数据库，注意这里先不指定需要连接的数据库名。
mysql -h 127.0.0.1 -P 3306 -u root -p"123456"
2.2 确保当前数据库为空
查看当前所在的数据库，确保该值为空。
select database();
2.3 查看此时变量character_set_connection对应的编码值
在确保database()的值为空的前提下查看character_set_connection对应的编码值。
show variables where Variable_name = 'character_set_connection';
变量character_set_connection对应的编码值为gbk，后面的配置需要用到此参数。
2.4 创建数据库
在mysql中创建一个名为mydatabase的数据库默认编码为UTF8供pymysql连接。
DROP DATABASE IF EXISTS mydatabase;CREATE DATABASE IF NOT EXISTS mydatabase DEFAULT CHARSET UTF8;
3 python端配置3.1 python代码思路
① 先用正则表达式对测试文本数据customer.txt清洗；
② 将清洗的结果保存为customer.csv；
③ Pandas读取customer.csv文件得到数据帧df；
④ 创建mysql数据引擎并将数据帧df写入到数据库mydatabase中保存为表customer
3.2 python代码源码
函数参数：
① tablename：文本文件名→mysql数据表名
② date_times：数据中需要转换为日期类型的数据对应的字段名列表。
注意事项：
确保26行的charset值与2.3看到的变量character_set_connection对应的编码值一致。
import reimport pandas as pdfrom sqlalchemy import create_engine# 准备数据def ready_datas(tablename:str,date_times:[str]=''): # 处理文本数据 with open("./datas/{}.txt".format(tablename), "r", encoding="utf-8") as file: text = file.read() text = re.sub(r"[ ]*[\|][ ]*", ",", text) text = text.split('\n')[1]+'\n'+'\n'.join(text.split('\n')[3:-1]) text = text.replace(",\n,", "\n")[1:-1] # 转换为csv文件 with open("./datas/{}.csv".format(tablename), "w", encoding="utf-8") as file: file.write(text) # 转换为datafram数据 df = pd.read_csv("./datas/{}.csv".format(tablename),encoding='utf-8') print('dataframe {}:success'.format(tablename)) # 数据覆盖写入mysql if date_times != '': for date_time in date_times: df = df.astype({date_time:"datetime64[ns]"}) # df[date_time] = pd.to_datetime(df[date_time]) # print(df.dtypes) engine = create_engine('mysql+pymysql://root:123456@localhost/mydatabase?charset=GBK') df.to_sql(name='{}'.format(tablename),index=None,con=engine,if_exists='replace') print('table {}:success'.format(tablename)) return dfcustomer = ready_datas('customer',date_times=['visited_on'])代码运行前如图所示，datas文件夹中仅有customer.txt文件。
代码运行后如图所示，datas文件夹下生成一个customer.csv的文件。
4 结果验证
打开mysql并连接创建好的数据库mydatabase。
mysql -h 127.0.0.1 -P 3306 -u root -p"123456" mydatabase;
使用mysql的dql语句查看生成的数据表customer的数据与表格文件customer.csv的数据是否一致。
select * from customer limit 3;
查看结果表明数据确实一致。