当前位置:首页>python>Python处理CSV文件的20个技巧:从中文乱码到GB级大文件,复制即用

Python处理CSV文件的20个技巧:从中文乱码到GB级大文件,复制即用

  • 2026-09-29 14:27:57
Python处理CSV文件的20个技巧:从中文乱码到GB级大文件,复制即用

CSV 看着简单,真处理起来坑也不少:中文乱码、几百万行读不动、特殊字符导致列错位、日期格式五花八门…… 本文整理了 20 个实战中最常用的技巧,csv 标准库 + pandas 双路线,每个都配可运行代码,复制就能用。

两套工具简述

场景
推荐
理由
超大文件流式读取、零依赖脚本
内置 csv 库
无需安装、内存友好
数据筛选、统计、清洗、合并、报表
pandas
一行完成复杂运算
导出给 Excel 查看
pandas + utf-8-sig
中文不乱码

环境准备:

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)

技巧清单

技巧
关键代码
应用场景
1. csv 逐行读取
csv.reader(f)
小型表格、零依赖
2. DictReader
row['name']
按列名取值
3. 写入/追加
csv.writer
 + newline=''
新建或追加 CSV
4. DictWriter
csv.DictWriter
字典数据写入
5. 引号字段
quoting=QUOTE_MINIMAL
含逗号、换行字段
6. 流式读取
yield
 生成器
GB 级大文件
7. 中文乱码
encoding='utf-8-sig'
Excel 导出的 CSV
8. 跳过脏行
skiprowsskipfooteron_bad_lines
系统导出带说明
9. 条件筛选
df[df['金额'] > 10000]
数据过滤
10. 分块读取
chunksize=10000
大文件处理
11. 列重命名
df.rename(columns={...})
表头中文化
12. 类型转换
pd.to_datetime().astype()
日期数字处理
13. 缺失值
fillna()dropna()interpolate()
数据清洗
14. 去重异常
drop_duplicates()
数据清洗
15. 分组统计
df.groupby('地区')['金额'].sum()
数据聚合
16. 合并多文件
pd.concat([...])
月报合并年报
17. 表关联
df.merge(..., on=..., how=...)
SQL JOIN
18. 拆分导出
df[df['city']==c].to_csv()
按值分文件
19. 导出 Excel
to_excel()
 + openpyxl
交付带格式报告
20. 特殊分隔符
sep='\t'
 `sep='
'`


作者简介:码上工坊,探索用编程为己赋能,定期分享编程知识和项目实战经验。持续学习、适应变化、记录点滴、复盘反思、成长进步。

重要提示:本文主要是记录自己的学习与实践过程,所提内容或者观点仅代表个人意见,只是我以为的,不代表完全正确,欢迎交流讨论。

最新文章

随机文章