Python Pandas数据输入输出全流程指南

1. 引言

数据输入输出(I/O)是数据分析工作流中最基础也是最重要的环节之一。Pandas提供了丰富的数据读写接口，支持从各种文件格式和数据库中加载数据，以及将处理后的数据保存到不同存储系统中。本文将全面介绍Pandas的数据I/O功能，包括常见文件格式解析、数据库交互、大数据处理技巧和高效存储格式。

2. 文件格式解析

2.1 CSV文件读写

import pandas as pd# 读取CSV文件
df = pd.read_csv('data.csv', encoding='utf-8', sep=',', header=0)# 查看前5行
print(df.head())# 写入CSV文件
df.to_csv('output.csv', index=False, encoding='utf-8')# 参数说明：
# - encoding: 文件编码(常用utf-8, gbk等)
# - sep: 分隔符(默认',')
# - header: 指定作为列名的行号(默认0，即第一行)
# - index: 是否写入行索引(默认True)

应用场景：CSV是最通用的数据交换格式，适合中小型数据集。

2.2 Excel文件读写

# 读取Excel文件
excel_data = pd.read_excel('data.xlsx', sheet_name='Sheet1')# 查看数据
print(excel_data.head())# 写入Excel文件
excel_data.to_excel('output.xlsx', sheet_name='Results', index=False)# 读取多个sheet
with pd.ExcelFile('data.xlsx') as xls:df1 = pd.read_excel(xls, 'Sheet1')df2 = pd.read_excel(xls, 'Sheet2')# 写入多个sheet
with pd.ExcelWriter('output_multi.xlsx') as writer:df1.to_excel(writer, sheet_name='Data1')df2.to_excel(writer, sheet_name='Data2')

注意事项：处理Excel文件需要安装openpyxl或xlrd库。

2.3 JSON文件读写

# 读取JSON文件
json_data = pd.read_json('data.json', orient='records')# 查看数据
print(json_data.head())# 写入JSON文件
json_data.to_json('output.json', orient='records', indent=2)# 参数说明：
# - orient: JSON格式(records, columns, index等)
# - indent: 缩进空格数(美化输出)# 从API获取JSON数据
import requests
url = 'https://api.example.com/data'
response = requests.get(url)
api_data = pd.read_json(response.text)

应用场景：Web API交互、半结构化数据存储。

2.4 其他文件格式

# Parquet文件(列式存储)
df = pd.read_parquet('data.parquet')
df.to_parquet('output.parquet')# HTML表格(读取网页中的表格)
html_tables = pd.read_html('https://example.com/table.html')# Pickle文件(Python对象序列化)
df.to_pickle('data.pkl')
df = pd.read_pickle('data.pkl')

3. 数据库交互

3.1 使用SQLAlchemy连接数据库

from sqlalchemy import create_engine# 创建数据库连接引擎(MySQL示例)
# 格式: dialect+driver://username:password@host:port/database
engine = create_engine('mysql+pymysql://user:password@localhost:3306/mydb')# 读取数据到DataFrame
query = "SELECT * FROM customers WHERE age > 30"
df = pd.read_sql(query, engine)# 写入数据到数据库
df.to_sql('new_table', engine, if_exists='replace', index=False)# 参数说明：
# - if_exists: 表存在时的行为(fail, replace, append)
# - index: 是否将索引作为列写入

3.2 PostgreSQL交互示例

# PostgreSQL连接示例
pg_engine = create_engine('postgresql+psycopg2://user:password@localhost:5432/mydb')# 执行复杂查询
complex_query = """SELECT c.name, COUNT(o.id) as order_countFROM customers cLEFT JOIN orders o ON c.id = o.customer_idGROUP BY c.nameORDER BY order_count DESC
"""
result = pd.read_sql(complex_query, pg_engine)

3.3 分块读取大型表

# 分块读取大型表
chunk_size = 10000
chunks = pd.read_sql_table('large_table', engine, chunksize=chunk_size)for chunk in chunks:# 处理每个数据块processed_chunk = chunk[chunk['value'] > 100]# 保存处理结果或进一步分析processed_chunk.to_sql('processed_data', engine, if_exists='append', index=False)

优势：避免内存不足问题，适合处理超大型数据库表。

4. 大数据处理技巧

4.1 分块读取大型文件

# 分块读取大型CSV文件
chunk_size = 50000
chunks = pd.read_csv('large_file.csv', chunksize=chunk_size)result_list = []
for chunk in chunks:# 对每个数据块进行处理filtered_chunk = chunk[chunk['sales'] > 1000]result_list.append(filtered_chunk)# 合并所有处理结果
final_result = pd.concat(result_list)

应用场景：处理内存无法容纳的超大型文件。

4.2 指定数据类型优化内存

# 读取时指定数据类型减少内存使用
dtype_dict = {'id': 'int32','name': 'category','value': 'float32','flag': 'bool'
}df = pd.read_csv('large_data.csv', dtype=dtype_dict)
print(df.memory_usage(deep=True))

效果：可显著减少内存占用，特别是对于包含字符串的列。

5. 高效数据存储格式

5.1 HDF5格式

# 存储到HDF5文件
df.to_hdf('data.h5', key='df', mode='w')# 从HDF5读取
df = pd.read_hdf('data.h5', key='df')# 存储多个DataFrame
store = pd.HDFStore('dataset.h5')
store['df1'] = df1
store['df2'] = df2
store.close()# 读取特定数据集
with pd.HDFStore('dataset.h5') as store:df1 = store['df1']df2 = store.get('df2')

特点：支持高效压缩，适合存储大型科学数据集。

5.2 Feather格式

# 存储为Feather格式
df.to_feather('data.feather')# 读取Feather文件
df = pd.read_feather('data.feather')# 特点：
# - 读写速度极快
# - 支持跨语言(R/Python)
# - 不适合长期存储(格式可能变化)

5.3 性能对比

import timeformats = ['csv', 'parquet', 'hdf', 'feather']
times = {}for fmt in formats:start = time.time()if fmt == 'csv':df.to_csv('test.csv', index=False)elif fmt == 'parquet':df.to_parquet('test.parquet')elif fmt == 'hdf':df.to_hdf('test.h5', key='data', mode='w')elif fmt == 'feather':df.to_feather('test.feather')times[f'write_{fmt}'] = time.time() - start# 类似地测试读取时间...

6. 数据I/O最佳实践

1. 数据安全：
   • 读写操作使用try-except处理异常
   • 重要数据写入前先备份
   • 数据库操作使用事务

from sqlalchemy.exc import SQLAlchemyErrortry:df.to_sql('important_data', engine, if_exists='replace')engine.execute("COMMIT")
except SQLAlchemyError as e:print(f"Database error occurred: {e}")engine.execute("ROLLBACK")

2. 性能优化：

   • 大数据集使用分块处理
   • 合理选择存储格式
   • 指定数据类型减少内存使用

3. 数据验证：

   • 读取后检查行数、列数和数据类型
   • 验证关键统计量是否符合预期

# 数据验证示例
def validate_data(df):assert len(df) > 0, "数据为空"assert 'id' in df.columns, "缺少ID列"assert df['value'].isna().sum() == 0, "存在空值"return True

7. 总结

1. 文件格式支持：

   • 结构化数据：CSV、Excel
   • 半结构化数据：JSON
   • 高效二进制格式：Parquet、HDF5、Feather

2. 数据库交互：

   • 使用SQLAlchemy作为统一接口
   • 支持MySQL、PostgreSQL等主流数据库
   • 分块处理大型表数据

3. 大数据处理：

   • chunksize参数分块读取
   • 指定dtype减少内存占用
   • 使用高效二进制格式存储中间结果

4. 存储选择建议：

   • 快速读写：Feather
   • 长期存储：Parquet或HDF5
   • 数据交换：CSV或JSON

Pandas强大的I/O功能使其成为数据科学工作流中的核心工具。掌握这些数据输入输出技术，能够让你：

• 高效地从各种数据源获取数据
• 处理超大规模数据集
• 选择最适合的存储格式
• 构建稳健的数据处理管道

在实际项目中，应根据数据规模、性能需求和使用场景，灵活选择和组合这些I/O方法。