在Python的数据存储与处理领域，CSV、JSON和MySQL被广大开发者誉为“数据存储三剑客”。它们各自在不同的场景下发挥着重要作用，无论是简单的数据交换、轻量级的数据存储，还是复杂的关系型数据库管理，都能找到它们的身影。本文将详细介绍如何使用Python的csv模块进行CSV文件的读写操作，json模块进行JSON数据的序列化处理，以及pymysql库连接MySQL数据库，并通过一个完整的案例展示如何构建一个数据管道。
1. CSV模块读写操作
CSV（Comma-Separated Values）文件是一种简单的文本文件格式，用于存储表格数据。Python的csv模块提供了丰富的功能，可以方便地进行CSV文件的读写操作。
读取CSV文件
import csv
def read_out():
with open('my-checkin.csv', 'r') as _file:
_reader = csv.reader(_file)
for row in _reader:
print(row)
if __name__ == '__main__':
read_out()上述代码会打开名为data.csv的文件，并将其内容逐行读取为列表形式，然后打印出来。
写入CSV文件
import csv
import random
def write_in():
# 定义数据标题
_list = [['date', 'in', 'out']]
# 模拟一个月的考勤数据
for i in range(31):
_h = random.randint(1, 30) + 10
_oh = random.randint(1, 30) + 20
_list.append([
'2025-03-' + (f'0{i}' if i < 10 else f'{i}'),
'08:' + str(_h),
'18:' + str(_oh)
])
# 打开csv文件，并写入数据
with open('my-checkin.csv', 'w', newline='') as checkin:
_writer = csv.writer(checkin)
_writer.writerows(_list)
if __name__ == '__main__':
write_in()上述代码会创建名为output.csv的文件，并将data列表中的数据逐行写入文件。
2. JSON数据序列化处理
JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，易于人阅读和编写，同时也易于机器解析和生成。Python的json模块提供了对JSON数据的解析和生成功能。
序列化（将Python对象转换为JSON字符串）
import json
def to_json():
# 定义字典数据
data = {
"name": "张三",
"age": 30,
"is_employee": True,
"skills": ["Python", "Java", "C++"]
}
# 将字典转换为JSON格式的字符串
json_data = json.dumps(data, ensure_ascii=False, indent=4)
print(json_data)
if __name__ == '__main__':
to_json()上述代码会将Python字典data转换为格式化的JSON字符串，并打印出来。
反序列化（将JSON字符串转换为Python对象）
import json
def to_obj():
# 定义一个JSON格式的字符串
json_data = '{"name": "李四", "age": 25, "is_employee": false, "skills": ["JavaScript", "HTML", "CSS"]}'
# 将JSON字符串转换为Python字典
data = json.loads(json_data)
print(data)
print(type(data))
if __name__ == '__main__':
to_obj()上述代码会将JSON字符串json_data转换为Python字典data，并打印出来及其类型。
3. pymysql数据库连接
pymysql是Python中用于连接MySQL数据库的一个库，它提供了与MySQL数据库进行交互的接口。
安装pymysql
首先，你需要使用pip安装pymysql库：
pip install pymysql cryptography连接MySQL数据库并执行查询
import pymysql
# 建立与MySQL数据库的连接
conn = pymysql.connect(
host='localhost',
user='yourusername',
password='yourpassword',
database='yourdatabase'
)
try:
with conn.cursor() as cursor:
# 执行SQL查询
sql = "SELECT * FROM yourtable"
cursor.execute(sql)
# 获取查询结果
results = cursor.fetchall()
for row in results:
print(row)
finally:
# 关闭数据库连接
conn.close()上述代码会连接到本地的MySQL数据库，执行一个SQL查询，并打印查询结果。
4. 案例：构建完整数据管道
下面，我们将通过一个完整的案例，展示如何构建一个从CSV文件读取数据，经过处理（如数据清洗、转换等），然后存储到MySQL数据库中的数据管道。
数据管道流程
- 从CSV文件读取数据。
- 对数据进行清洗和转换。
- 将处理后的数据存储到MySQL数据库中。
示例代码
原始数据(house_info.csv)
总价,单位价格,楼房信息
277,16380元/平米,2002年建/板塔结合
256,29869元/平米,2009年建/塔楼
500,30031元/平米,2005年建/板楼
215,13699元/平米,2000年建/塔楼
125,15117元/平米,2003年建/板楼
305,39173元/平米,2012年建/塔楼
98,11829元/平米,1992年建/平房
325,17580元/平米,2000年建/塔楼
290,24593元/平米,2015年建/板塔结合
232,26135元/平米,未知年建/板塔结合
153,17035元/平米,2013年建/板塔结合
225,26163元/平米,2011年建/塔楼
102,12848元/平米,未知年建/塔楼
126,12000元/平米,1998年建/板楼
167,18490元/平米,2013年建/塔楼
205,25764元/平米,2017年建/塔楼
240,19486元/平米,2008年建/板塔结合
155,20893元/平米,2012年建/板塔结合从CSV文件读取数据(read_csv.py)
import csv
def read_csv(_filename):
try:
_result = []
with open(_filename, 'r') as file:
_reader = csv.reader(file)
next(_reader)
for line in _reader:
_result.append(line)
print('------------------------- read success')
return _result
except csv.Error as e:
print('------------------------- read error')
print(e)
return []对数据进行清洗和转换(clean_data.py)
import re
def is_num(_s):
return re.match('^\d*\.?\d*$', _s)
def clean_data(_list):
_result = []
try:
# 原数据
# ['279', '29531元/平米', '未知年建/板塔结合']
# ['279', '29531元/平米', '2005年建/板塔结合']
for _row in _list:
_s2 = re.sub('[^0-9]', '', _row[1])
_s3 = _row[2].split('/')
_result.append([
round(float(_row[0]) if is_num(_row[0]) else 0, 2),
0 if '' else int(_s2),
_s3[0],
_s3[1],
])
except Exception as e:
print('------------------------- clean error')
print(e)
print('------------------------- clean success')
return _result将处理后的数据存储到MySQL数据库中(store_data.py)
import pymysql
def store_data(data):
_conn = None
try:
_conn = pymysql.connect(
host='localhost',
user='python_demo',
database='python_demo',
password='kX#8+N4$@YQt@+EAOG&mj!*%jUvgtms&$'
)
with _conn.cursor() as cursor:
cursor.execute('''
CREATE TABLE IF NOT EXISTS csv_demo_table (
amount DECIMAL(10,2),
price INT,
build_year VARCHAR(40),
house_type VARCHAR(40)
)
''')
# 插入数据
sql = "INSERT INTO csv_demo_table (amount, price, build_year, house_type) VALUES (%s, %s, %s, %s)"
cursor.executemany(sql, data)
# 提交事务
_conn.commit()
except pymysql.Error as e:
print('------------------------- mysql error')
print(e)
finally:
if _conn:
_conn.close()
print('------------------------- store success')执行整体业务数据(main.py)
from clean_data import clean_data
from read_csv import read_csv
from store_data import store_data
if __name__ == '__main__':
# 调用读取方法获取csv数据
_list = read_csv('house_info.csv')
# 调用清洗方法获取结果集
_result = clean_data(_list)
# 存储数据到Mysql数据库中
store_data(_result)
print('------------------ finish')
说明
- read_csv函数：从指定的CSV文件读取数据，并跳过表头行。
- clean_data函数：对读取的数据进行清洗和转换，如将年龄字段转换为整数。
- store_data函数：将处理后的数据存储到MySQL数据库中。首先检查表是否存在，如果不存在则创建表，然后插入数据并提交事务。
- main函数：作为程序的入口，依次调用上述三个函数完成整个数据管道的流程。
通过以上步骤，我们成功构建了一个从CSV文件读取数据，经过处理，然后存储到MySQL数据库中的数据管道。这个案例展示了Python在数据处理和存储方面的强大能力。
关注我！！🫵 持续为你带来Nodejs相关内容。
