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 Caption
python –version
pip 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 --version
mysql -h 127.0.0.1 -P 3306 -u root -p"123456"
1.3 数据准备
运行以下python代码准备模拟数据用于测试。
import os
if 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 re
import pandas as pd
from 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 df
customer = 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;
查看结果表明数据确实一致。