CSV 看着简单,真处理起来坑也不少:中文乱码、几百万行读不动、特殊字符导致列错位、日期格式五花八门…… 本文整理了 20 个实战中最常用的技巧,csv 标准库 + pandas 双路线,每个都配可运行代码,复制就能用。
两套工具简述
环境准备:
pip install pandas openpyxl
csv 标准库(零依赖、轻量)
技巧1:csv.reader 逐行读取
最基础的读取方式,每行返回列表,按下标取值。适合几 MB 以内小型表格。
import csvwith open('data.csv', 'r', encoding='utf-8') as f: reader = csv.reader(f) header = next(reader) # 第一行是表头 print('表头:', header)for row in reader:# row = ['张三', '22', '北京'] name, age, city = row print(f'姓名:{name},年龄:{age}')
注意:csv.reader 读出的全是字符串,数字需手动 int() / float()。
技巧2:csv.DictReader 按列名读取
通过表头名称取值,不用记列下标,表头顺序调整后代码无需改。
import csvwith open('data.csv', 'r', encoding='utf-8') as f: reader = csv.DictReader(f) print('字段名:', reader.fieldnames)for row in reader:# 直接用列名取值 print('姓名:', row['name'], '城市:', row['city']) age = int(row['age']) # 数值转换
技巧3:csv.writer 写入与追加
新建、追加 CSV,**必加 newline=''**,否则 Windows 会多出空行。
import csvheader = ['姓名', '年龄', '城市']data = [ ['李四', 25, '上海'], ['王五', 30, '广州']]# 新建写入(mode='w')with open('output.csv', 'w', encoding='utf-8-sig', newline='') as f: writer = csv.writer(f) writer.writerow(header) # 写表头 writer.writerows(data) # 批量写数据# 追加写入(mode='a',不覆盖原数据)with open('output.csv', 'a', encoding='utf-8-sig', newline='') as f: writer = csv.writer(f) writer.writerow(['赵六', 28, '深圳'])
技巧4:csv.DictWriter 按字典写入
数据是字典列表时用 DictWriter,自动按 fieldnames 顺序写入。
import csvheader = ['name', 'age', 'city']rows = [ {'name': '小明', 'age': 21, 'city': '杭州'}, {'name': '小红', 'age': 23, 'city': '成都'}]with open('dict_out.csv', 'w', encoding='utf-8-sig', newline='') as f: writer = csv.DictWriter(f, fieldnames=header) writer.writeheader() # 自动写表头 writer.writerows(rows)
技巧5:正确处理带引号、含逗号、内换行的字段
字段里有逗号或换行符时,千万别用 line.split(','),会直接列错位。csv 库自动解析引号包裹的内容。
import csv# 写入:自动给含特殊字符的字段加引号data = [ ['Name', 'Description', 'Price'], ['Product A', 'Contains commas, and "quotes"', '100'], ['Product B', 'Line1\nLine2', '200']]with open('products.csv', 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f, quoting=csv.QUOTE_MINIMAL) # 仅必要时加引号 writer.writerows(data)# 读取:引号和换行被正确还原with open('products.csv', 'r', encoding='utf-8') as f:for row in csv.reader(f): print(row)
quoting 参数:QUOTE_MINIMAL(默认,推荐)、QUOTE_ALL(全加)、QUOTE_NONNUMERIC(非数字加)、QUOTE_NONE(慎用)。
技巧6:超大文件流式读取
GB 级文件不能一次性读入内存。用生成器逐行处理,内存占用几乎为零。
import csvdefprocess_big_csv(file_path):with open(file_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f)for count, row in enumerate(reader, 1):# 逐行处理,例如筛选年龄大于 25try:if int(row['age']) > 25:yield rowexcept (ValueError, KeyError):continueif count % 1000 == 0: print(f'已处理 {count} 行')# 边读边处理,内存恒定for item in process_big_csv('big_data.csv'):pass
pandas 高效处理(数据分析首选)
技巧7:解决中文乱码与 BOM 头
读 CSV 第一坑:中文表头变问号、第一列名变成 \ufeffcolumn。99% 是编码问题。
import pandas as pd# Excel 导出的 CSV 用 utf-8-sig(带 BOM,自动去除 \ufeff)df = pd.read_csv('data.csv', encoding='utf-8-sig')# 老版本 Excel 可能用 gbkdf = pd.read_csv('data.csv', encoding='gbk')# 兜底容错:先 utf-8 失败再 gbktry: df = pd.read_csv('data.csv', encoding='utf-8')except UnicodeDecodeError: df = pd.read_csv('data.csv', encoding='gbk')
确定编码:用记事本打开 CSV → 另存为 → 看编码选项,是什么编码 read_csv 就传什么。
技巧8:跳过空行、注释行、错误行
业务系统导出的 CSV 经常夹着表头说明、合计行、注释、列数错位行。
# 跳过前 3 行(表头说明)df = pd.read_csv('data.csv', skiprows=3)# 跳过指定行(0索引)df = pd.read_csv('data.csv', skiprows=[1, 5, 10])# 跳过底部 2 行(合计行),需 engine='python'df = pd.read_csv('data.csv', skipfooter=2, engine='python')# 自动跳过空行df = pd.read_csv('data.csv', skip_blank_lines=True)# 忽略 # 开头的注释行df = pd.read_csv('data.csv', comment='#')# 跳过列数不一致的错误行df = pd.read_csv('data.csv', on_bad_lines='skip')
技巧9:按条件筛选行
Excel 筛选几万行就卡,pandas 几百万行秒开。
df = pd.read_csv('sales.csv', encoding='utf-8-sig')# 单条件big = df[df['销售额'] > 10000]# 多条件 AND / ORresult = df[(df['销售额'] > 10000) & (df['地区'] == '华东')]result = df[(df['地区'] == '华东') | (df['地区'] == '华南')]# IN 查询result = df[df['地区'].isin(['华东', '华南', '华北'])]# 字符串包含result = df[df['商品名'].str.contains('手机', na=False)]# 取反result = df[~df['地区'].isin(['华东', '华南'])]
技巧10:分块读取大文件
CSV 超过 1GB,read_csv 一次性加载会撑爆内存。用 chunksize 分块。
# 方式1:逐块统计聚合total = 0for chunk in pd.read_csv('big_data.csv', chunksize=10000): total += chunk['金额'].sum()print(f'总金额: {total}')# 方式2:筛选后合并result = []for chunk in pd.read_csv('big_data.csv', chunksize=10000): result.append(chunk[chunk['金额'] > 1000])df = pd.concat(result, ignore_index=True)
进阶:只读需要的列,进一步省内存。
df = pd.read_csv('big.csv', usecols=['日期', '金额', '地区'])
技巧11:列重命名三种写法
CSV 表头是英文、缩写、乱码,读进来第一件事改成中文。
# 方式1:rename(推荐,最清晰)df = df.rename(columns={'date': '日期', 'amount': '金额', 'region': '地区'})# 方式2:直接改 columns(按位置,需列数一致)df.columns = ['日期', '金额', '地区', '备注']# 方式3:读取时指定(原表头无用)df = pd.read_csv('data.csv', names=['日期', '金额', '地区'], header=0)
技巧12:类型转换,日期数字不再乱
CSV 读进来全是字符串?日期没法排序?数字带千分位?一招解决。
# 数字列带千分位逗号df['金额'] = df['金额'].str.replace(',', '', regex=False).astype(float)# 日期列(格式统一时指定 format 最快)df['日期'] = pd.to_datetime(df['日期'], format='%Y-%m-%d')# 多种日期格式混合(2024-01-01 / 2024/01/02 / 2024.01.03)df['日期'] = pd.to_datetime(df['日期'], format='mixed', errors='coerce')# 整数列(空值变 NaN 无法直接转 int,需先 fillna)df['数量'] = df['数量'].fillna(0).astype(int)# 分类列(省内存、提速)df['地区'] = df['地区'].astype('category')
读取时直接指定类型(避免后续转换,手机号/编号务必设 str 防科学计数法):
df = pd.read_csv('data.csv', dtype={'age': int, 'id': str, 'phone': str})
技巧13:缺失值处理
CSV 中常有空单元格,灵活处理是清洗关键。
# 读取时指定缺失值标记df = pd.read_csv('data.csv', na_values=['', 'NULL', 'N/A', 'null'])# 不同列用不同方式填充df['age'] = df['age'].fillna(df['age'].median()) # 数值列用中位数df['city'] = df['city'].fillna('未知') # 分类列填默认值df['score'] = df['score'].interpolate() # 时间序列用插值# 删除缺失过多的行(至少保留 n-2 个非空值)df = df.dropna(thresh=len(df.columns) - 2)# 仅指定列空值才删除df = df.dropna(subset=['name', 'age'])
技巧14:数据清洗——重复行与异常值
# 删除完全重复行df = df.drop_duplicates()# 按指定列去重,保留第一条df = df.drop_duplicates(subset=['name'], keep='first')# 过滤异常数字df = df[(df['age'] > 0) & (df['age'] < 120)]# 字符串去空格df['name'] = df['name'].str.strip()
技巧15:分组统计,Excel 透视表的 Python 版
分组聚合是数据分析最高频操作,pandas 一行搞定。
# 按地区求销售额总和result = df.groupby('地区')['金额'].sum()# 多个聚合函数result = df.groupby('地区')['金额'].agg(['sum', 'mean', 'count', 'max'])# 重命名聚合列result = df.groupby('地区')['金额'].agg( 总额='sum', 均值='mean', 笔数='count')# 按多列分组result = df.groupby(['地区', '商品'])['金额'].sum()# 分组后重置索引(变 DataFrame 而非 Series)result = df.groupby('地区')['金额'].sum().reset_index()
输出效果:
地区 总额 均值 笔数0 华东 125000 12500.0 101 华南 98000 9800.0 102 华北 76000 7600.0 10
技巧16:合并多个 CSV(按行拼接)
每月一份销售表,年底合并成年报,一个循环搞定。
from pathlib import Pathimport pandas as pd# 方式1:glob 匹配import globfiles = glob.glob('sales_*.csv')df = pd.concat([pd.read_csv(f, encoding='utf-8-sig') for f in files], ignore_index=True)# 方式2:pathlib(更优雅)files = sorted(Path('.').glob('sales_*.csv'))df = pd.concat([pd.read_csv(f, encoding='utf-8-sig') for f in files], ignore_index=True)# 加来源列,方便追溯dfs = []for f in files: tmp = pd.read_csv(f, encoding='utf-8-sig') tmp['来源文件'] = f.name dfs.append(tmp)df = pd.concat(dfs, ignore_index=True)
技巧17:表关联(按列 JOIN)
类似 SQL 的 JOIN,把两张表按某列关联起来。
df_orders = pd.read_csv('orders.csv')df_customers = pd.read_csv('customers.csv')# 内连接(只保留两表都有的 customer_id)merged = df_orders.merge(df_customers, on='customer_id', how='inner')# 左连接(保留左表全部行)merged = df_orders.merge(df_customers, on='customer_id', how='left')# 列名不同时用 left_on / right_onmerged = df_orders.merge(df_customers, left_on='cust_id', right_on='customer_id', how='inner')
技巧18:按条件拆分导出多个 CSV
把一张总表按某列的值拆成多个文件,例如按城市拆分。
import osos.makedirs('city_data', exist_ok=True)df = pd.read_csv('user.csv', encoding='utf-8-sig')for city in df['city'].unique(): city_data = df[df['city'] == city] city_data.to_csv(f'city_data/{city}.csv', index=False, encoding='utf-8-sig') print(f'导出 {city}.csv:{len(city_data)} 行')
技巧19:导出带格式 Excel
CSV 没有格式、列宽乱、数字没千分位。导出 Excel 自动调整。
from openpyxl import load_workbookfrom openpyxl.styles import Font, PatternFilldf.to_excel('output.xlsx', index=False, engine='openpyxl')wb = load_workbook('output.xlsx')ws = wb.active# 表头加粗 + 蓝底白字for cell in ws[1]: cell.font = Font(bold=True, color='FFFFFF') cell.fill = PatternFill('solid', fgColor='4472C4')# 自动列宽for col in ws.columns: max_len = max(len(str(cell.value)) for cell in col) ws.column_dimensions[col[0].column_letter].width = max_len + 2wb.save('output.xlsx')
技巧20:处理特殊分隔符
不是所有 CSV 都用逗号,制表符、分号、竖线很常见。读错分隔符整行变成一列。
# 制表符分隔(.tsv)df = pd.read_csv('data.tsv', sep='\t')# 分号分隔(欧洲 Excel 默认)df = pd.read_csv('data.csv', sep=';')# 竖线分隔df = pd.read_csv('data.csv', sep='|')# 多个空格分隔df = pd.read_csv('data.csv', sep=r'\s+', engine='python')# 自动检测分隔符(sniffer)df = pd.read_csv('data.csv', engine='python', sep=None)
字段内含逗号(用引号包围)pandas 默认处理,无需额外操作:
姓名,城市,备注张三,"北京,朝阳","喜欢足球,篮球"
高阶工具
DuckDB 直接查 CSV,速度比 pandas 快 5-10 倍
文件超过 1GB,pandas 还是慢?试试 DuckDB,SQL 语法直接查 CSV,不加载到内存。
import duckdbresult = duckdb.query(""" SELECT 地区, SUM(金额) as 总额, COUNT(*) as 笔数 FROM 'big_data.csv' WHERE 金额 > 10000 GROUP BY 地区 ORDER BY 总额 DESC""").to_df()print(result)
DuckDB 优势:SQL 直接查文件、自动识别表头和类型、支持多文件 JOIN。
pyarrow 引擎加速 pandas
# 读取速度提升 2-5 倍(需 pip install pyarrow)df = pd.read_csv('large.csv', engine='pyarrow')# 一次性推断类型,避免分块警告df = pd.read_csv('large.csv', low_memory=False)
技巧清单
| | |
|---|
| csv.reader(f) | |
| row['name'] | |
| csv.writer | |
| csv.DictWriter | |
| quoting=QUOTE_MINIMAL | |
| yield | |
| encoding='utf-8-sig' | |
| skiprowsskipfooteron_bad_lines | |
| df[df['金额'] > 10000] | |
| chunksize=10000 | |
| df.rename(columns={...}) | |
| pd.to_datetime().astype() | |
| fillna()dropna()interpolate() | |
| drop_duplicates() | |
| df.groupby('地区')['金额'].sum() | |
| pd.concat([...]) | |
| df.merge(..., on=..., how=...) | |
| df[df['city']==c].to_csv() | |
| to_excel() | |
| sep='\t' | |