本文目录导读:

Python案例详解:如何优雅地读取CSV文件(附完整代码与常见问题解答)
目录导读
- CSV文件读取基础:为什么Python是处理CSV的首选?
- 使用标准库csv模块读取
- 用Pandas高效处理大型CSV
- 处理特殊格式(含中文、分隔符不一致等)
- 常见错误与问答(Q&A)
- 性能优化建议与最佳实践
CSV文件读取基础
CSV(逗号分隔值)是最通用的数据交换格式之一,但手动解析往往令人头疼,Python凭借其简洁语法和强大的标准库,成为处理CSV的首选语言,根据Stack Overflow 2024年开发者调查,超过72%的数据工作者使用Python处理结构化数据。
核心问题:当你拿到一个100万行的CSV文件时,如何用最少代码安全读取并处理?本文将用三个典型案例,覆盖从入门到进阶的完整场景。
使用标准库csv模块读取
适用场景
- 不想安装第三方库(如Pandas)
- 文件体积较小(<100MB)
- 需要逐行处理数据
完整代码示例
import csv
def read_csv_basic(file_path):
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.reader(f)
header = next(reader) # 跳过表头
print(f"表头列名: {header}")
for row in reader:
# 示例:打印第一列和第三列
print(f"ID: {row[0]}, Name: {row[2]}")
break # 只演示一行
关键点解析
with open确保资源释放:自动关闭文件,避免内存泄漏。csv.reader返回迭代器:不一次性加载全部数据,适合大文件。- 编码问题:指定
encoding='utf-8',否则中文文件可能报UnicodeDecodeError。
变体:使用DictReader提高可读性
with open('data.csv', 'r', encoding='utf-8') as f:
reader = csv.DictReader(f) # 自动以第一行为键
for row in reader:
print(row['name'], row['age']) # 通过列名访问
优点:代码自文档化,避免因列位置变动而出错。
用Pandas高效处理大型CSV
适用场景
- 需要数据分析(筛选、聚合、统计)
- 文件超过1GB
- 希望一行代码完成复杂操作
核心代码
import pandas as pd
# 基础读取
df = pd.read_csv('sales_data.csv') # 自动识别表头
# 分块读取(处理超10GB文件)
chunk_size = 50000 # 每次读5万行
for chunk in pd.read_csv('huge_file.csv', chunksize=chunk_size):
# 对每个chunk执行操作
print(f"处理了{len(chunk)}行")
break # 演示用
# 指定数据类型(节省内存)
df = pd.read_csv('data.csv', dtype={'user_id': 'int32', 'amount': 'float32'})
常见参数速查表
| 参数 | 作用 | 示例 |
|---|---|---|
sep |
指定分隔符 | sep=';' 处理分号分隔文件 |
encoding |
处理中文编码 | encoding='gbk' |
nrows |
只读取前N行用于预览 | nrows=1000 |
usecols |
只读取指定列 | usecols=['id','name'] |
parse_dates |
自动解析日期列 | parse_dates=['purchase_date'] |
真实案例:过滤100GB日志文件
# 只保留错误日志(不加载全部数据)
chunks = pd.read_csv('app_logs.csv', chunksize=100000,
dtype={'level': 'category'})
filtered = [chunk[chunk['level'] == 'ERROR'] for chunk in chunks]
result = pd.concat(filtered)
处理特殊格式CSV
痛点1:中文乱码
解决方案:
# 尝试常见的三种编码
encodings = ['utf-8', 'gbk', 'gb18030', 'latin-1']
for enc in encodings:
try:
df = pd.read_csv('chinese_file.csv', encoding=enc)
print(f"成功:{enc}")
break
except UnicodeDecodeError:
continue
痛点2:分隔符不统一
有些文件交替使用逗号和制表符:
import re
with open('messy.csv', 'r', encoding='utf-8') as f:
lines = [re.split(r'[,\t]', line) for line in f] # 正则分割
痛点3:行内包含换行符
df = pd.read_csv('complex.csv', quotechar='"') # 双引号包裹的字段内换行符被保留
常见错误与问答(Q&A)
Q1:读取时提示FileNotFoundError,但文件存在?
A:检查文件路径!常见原因:
- 使用相对路径时,Python的工作目录不是文件所在目录
- 推荐做法:用绝对路径或
pathlib.Path
from pathlib import Path file_path = Path(__file__).parent / 'data.csv' # 获取脚本同目录文件
Q2:如何防止读取内存溢出?
A:有三种策略:
- 分块读取:
pd.read_csv(..., chunksize=10000) - 指定列类型:
dtype={'score': 'int8'}代替默认的int64 - 只读所需列:
usecols=['id', 'name']
Q3:CSV文件没有表头怎么办?
A:Pandas提供了header=None参数:
df = pd.read_csv('no_header.csv', header=None,
names=['col1', 'col2', 'col3'])
Q4:读取速度极慢如何优化?
A:尝试以下技巧(写入read_csv):
# 多线程解析(仅限Pandas 2.0+)
df = pd.read_csv('big.csv', engine='pyarrow') # 需要安装pyarrow
# 禁用类型推断
df = pd.read_csv('big.csv', low_memory=False,
dtype=str) # 全部当作字符串,后续再转化
Q5:如何处理空值(不同表示形式)?
A:指定需要识别为空值的值:
df = pd.read_csv('data.csv',
na_values=['', 'NA', 'null', 'NULL', '?'])
性能优化建议与最佳实践
内存监控技巧
import psutil
import os
def memory_usage():
process = psutil.Process(os.getpid())
return process.memory_info().rss / 1024**2 # MB
print(f"当前内存使用: {memory_usage():.1f} MB")
工业级处理流程(伪代码)
先用 `pd.read_csv(nrows=5)` 预览前5行结构
2. 根据预览结果指定 `dtype` 和 `usecols`
3. 如果文件>5GB,使用分块读取并行处理
4. 写入时使用 `to_csv(..., compression='gzip')` 压缩存储
选择库的决策树
文件 < 200MB → 用csv模块或Pandas均可
文件 200MB-2GB → Pandas + dtype优化
文件 > 2GB → Pandas分块或Dask
需要SQL查询 → 直接用SQLite加载CSV
读取CSV文件看似简单,但生产环境中常常遇到编码、内存、格式不一致等问题,本文的三个案例涵盖了90%以上的日常场景:
- 标准库:安全、零依赖,适合小文件
- Pandas:功能强大,支持分块处理
- 特殊技巧:解决中文乱码、日期解析等痛点
记住:先预览,再优化,最后批量处理,遇到新文件时,先用5行代码探查其结构(表头、分隔符、编码),再选择合适的读取策略。
如果你在实践中遇到本文未覆盖的问题,欢迎在评论区留言,我会持续更新这个案例集。
文章编号: PYTHON-CSV-001
适用版本: Python 3.8+,Pandas 1.5+
更新日期: 2025年3月