当前位置：首页 > news >正文

Python3 模块精讲：csv --读写 CSV 表格文件（完整版・超多实战代码）

news 2026/4/23 0:05:48

📝 本章学习目标：本章聚焦 Python 数据处理核心技能，帮助读者从零到一掌握csv 标准库的完整用法、工程实践与避坑指南。通过本章学习，你将全面掌握「Python3 csv 模块：读写 CSV 表格文件」全场景解决方案，实现 CSV 文件安全读写、复杂格式兼容、大批量数据处理、异常兼容与工程化落地，成为自动化办公、数据清洗、报表生成领域的实战型开发者。

一、引言：为什么 csv 模块是 Python 数据处理必备

在数据分析、自动化办公、爬虫存储、接口对接、日志处理等几乎所有 Python 工程场景中，CSV（Comma-Separated Values）都是最轻量化、最通用、跨平台无门槛的数据交换格式。它不需要安装 Office、不需要依赖数据库、不需要复杂驱动，文本编辑器即可打开，Excel/WPS/Numbers/MySQL/PostgreSQL/Spark 全生态兼容。

1.1 背景与意义

💡 核心认知：csv 模块是 Python 处理表格数据的 “原生基石”，零安装、零依赖、稳定可靠、兼容 RFC 4180 标准。

内置标准库：Python3 全版本自带，无需pip install，开箱即用
标准兼容：严格遵循国际 CSV 规范，自动处理引号、逗号、换行、空值
场景全覆盖：自动化报表、用户数据导出、日志解析、批量录入、配置读取
安全稳定：不会像open()直接读写那样出现格式错乱、编码崩溃、数据丢失

1.2 本章结构概览

plaintext

📊 概念解析 → 核心API → 读取CSV → 写入CSV → 高级用法 → 实战案例（超多） → 最佳实践 → 常见问题 → 总结展望

二、核心概念解析

2.1 基本定义

概念一：CSV 格式标准

CSV 是一种以纯文本存储表格数据的格式，核心规则：

每行代表一条记录（一行数据）
字段之间用分隔符（默认逗号,）隔开
包含逗号、引号、换行符的字段必须用双引号包裹
首行常作为表头（Header）
支持空值、空行、自定义分隔符

概念二：csv 模块核心能力

Python3 内置csv模块提供四大核心能力：

按行读取：逐行读取 CSV，低内存占用，支持超大文件
按行写入：逐行 / 批量写入，自动转义特殊字符
字典读写：以key-value方式操作，更符合编程直觉
方言控制：自定义分隔符、引号、换行符，兼容所有 CSV 变体

三、核心 API 与基础用法

3.1 模块导入

python

运行

# 导入Python内置csv模块，无需安装 import csv

3.2 基础读取：csv.reader（返回列表）

python

运行

# 基础读取示例：按行读取CSV，返回每行数据列表 import csv # 打开文件：utf-8编码，newline=""避免空行问题 with open("data.csv", "r", encoding="utf-8", newline="") as f: # 创建csv读取器对象 reader = csv.reader(f) # 读取第一行作为表头 header = next(reader) print("表头：", header) # 循环读取每一行数据 for row in reader: print("数据行：", row)

3.3 基础写入：csv.writer（写入列表）

python

运行

# 基础写入示例：将列表数据写入CSV文件 import csv # 定义要写入的数据（二维列表） rows = [ ["姓名", "年龄", "城市", "备注"], ["张三", 23, "北京", "擅长Python、Java"], ["李四", 25, "上海", "擅长,数据分析"], ["王五", 27, "广州", "擅长\"AI开发\""] ] # 打开文件，写入模式 with open("output.csv", "w", encoding="utf-8", newline="") as f: # 创建csv写入器 writer = csv.writer(f) # 批量写入所有行 writer.writerows(rows)

3.4 字典读取：csv.DictReader

python

运行

# 字典读取：按字段名读取，更直观 import csv with open("data.csv", "r", encoding="utf-8", newline="") as f: # 创建字典读取器，自动把第一行作为key reader = csv.DictReader(f) # 遍历每一行（每行是字典） for row in reader: print(f"姓名：{row['姓名']}, 年龄：{row['年龄']}, 城市：{row['城市']}")

3.5 字典写入：csv.DictWriter

python

运行

# 字典写入：直接写入字典数据，适合业务数据 import csv # 定义表头字段（必须与字典key一致） fieldnames = ["姓名", "年龄", "城市", "备注"] # 待写入数据（列表套字典） rows = [ {"姓名": "张三", "年龄": 23, "城市": "北京", "备注": "Python开发"}, {"姓名": "李四", "年龄": 25, "城市": "上海", "备注": "数据分析师"} ] with open("dict_output.csv", "w", encoding="utf-8", newline="") as f: # 创建字典写入器，指定字段名 writer = csv.DictWriter(f, fieldnames=fieldnames) # 写入表头 writer.writeheader() # 批量写入数据 writer.writerows(rows)

四、高级用法

4.1 自定义方言（处理特殊分隔符）

python

运行

# 自定义方言：处理竖线|分隔、制表符分隔等特殊CSV import csv # 注册自定义方言：分隔符为| csv.register_dialect("mydialect", delimiter="|", quotechar='"', skipinitialspace=True) # 使用自定义方言读取 with open("pipe.csv", "r", encoding="utf-8") as f: reader = csv.reader(f, dialect="mydialect") for row in reader: print(row)

4.2 四种引用模式

python

运行

# 所有字段强制加引号，兼容性最强 import csv rows = [["姓名","年龄"],["张三",23]] with open("quote_all.csv", "w", encoding="utf-8", newline="") as f: writer = csv.writer(f, quoting=csv.QUOTE_ALL) writer.writerows(rows)

4.3 超大文件逐行处理

python

运行

# 处理GB级超大CSV，低内存占用，不卡顿 import csv def process_big_file(file_path): count = 0 # 逐行读取，不一次性加载全部内容 with open(file_path, "r", encoding="utf-8") as f: reader = csv.reader(f) next(reader) # 跳过表头 for row in reader: count += 1 # 每10万行打印一次进度 if count % 100000 == 0: print(f"已处理 {count} 行") process_big_file("large_data.csv")

4.4 Windows 乱码解决方案

python

运行

# Windows Excel打开不乱码：使用utf-8-sig编码 import csv rows = [["姓名","年龄"],["张三",23]] # utf-8-sig 带BOM，Excel/WPS完美识别 with open("win_safe.csv", "w", encoding="utf-8-sig", newline="") as f: writer = csv.writer(f) writer.writerows(rows)

五、海量实战代码（高频场景・全注释）

实战 1：CSV 数据筛选（筛选年龄大于 25 的记录）

python

运行

import csv # 存储筛选结果 result = [] # 读取原文件 with open("data.csv", "r", encoding="utf-8", newline="") as f: reader = csv.DictReader(f) # 遍历每一行，按条件筛选 for row in reader: if int(row["年龄"]) > 25: result.append(row) # 将筛选结果写入新CSV with open("filtered.csv", "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=reader.fieldnames) writer.writeheader() writer.writerows(result)

实战 2：CSV 数据统计（计算平均年龄、人数）

python

运行

import csv # 初始化统计变量 total_age = 0 count = 0 with open("data.csv", "r", encoding="utf-8", newline="") as f: reader = csv.DictReader(f) for row in reader: # 累加年龄 total_age += int(row["年龄"]) # 计数+1 count += 1 # 输出结果 print(f"总人数：{count}") print(f"平均年龄：{total_age/count:.2f}")

实战 3：CSV 数据去重（按姓名去重）

python

运行

import csv # 用集合记录已存在的姓名，用于去重 names = set() rows = [] with open("data.csv", "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: # 如果姓名不在集合中，保留该行 if row["姓名"] not in names: names.add(row["姓名"]) rows.append(row) # 写入去重后的数据 with open("unique.csv", "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=reader.fieldnames) writer.writeheader() writer.writerows(rows)

实战 4：多 CSV 文件合并（批量合并到一个文件）

python

运行

import csv import os # 要合并的文件列表 csv_list = ["file1.csv", "file2.csv", "file3.csv"] # 统一表头 fieldnames = ["姓名", "年龄", "城市"] # 打开最终合并文件 with open("merged.csv", "w", encoding="utf-8", newline="") as out_f: writer = csv.DictWriter(out_f, fieldnames=fieldnames) writer.writeheader() # 遍历每个文件并读取 for file in csv_list: if os.path.exists(file): with open(file, "r", encoding="utf-8") as in_f: reader = csv.DictReader(in_f) # 写入到合并文件 writer.writerows(reader)

实战 5：一个 CSV 拆分成多个文件（按城市拆分）

python

运行

import csv from collections import defaultdict # 按城市分组存储数据 data = defaultdict(list) fieldnames = ["姓名", "年龄", "城市"] with open("data.csv", "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: city = row["城市"] # 按城市分组 data[city].append(row) # 每组生成一个CSV文件 for city, rows in data.items(): filename = f"{city}.csv" with open(filename, "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows)

实战 6：JSON 数据转 CSV

python

运行

import csv import json # 模拟JSON数据 json_data = [ {"name": "张三", "age": 23, "city": "北京"}, {"name": "李四", "age": 25, "city": "上海"} ] # 指定字段名 fieldnames = ["name", "age", "city"] # 直接写入字典数据 with open("json2csv.csv", "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(json_data)

实战 7：列表数据导出 CSV（爬虫专用）

python

运行

import csv # 模拟爬虫采集到的数据 data = [ ["标题", "链接", "阅读量"], ["Python入门", "https://xxx.com", 1000], ["csv教程", "https://xxx.com", 2000] ] # 快速导出为CSV with open("spider.csv", "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerows(data)

实战 8：读取无表头的 CSV（自定义表头）

python

运行

import csv # 读取没有表头的CSV文件 with open("no_header.csv", "r", encoding="utf-8") as f: reader = csv.reader(f) # 自定义表头 fieldnames = ["姓名", "年龄", "城市"] # 将每行列表转为字典 rows = [dict(zip(fieldnames, row)) for row in reader] print(rows)

实战 9：CSV 追加写入（不覆盖原有数据）

python

运行

import csv # 要追加的新数据 new_row = ["赵六", 30, "深圳", "运维工程师"] # 打开模式改为 a = append 追加 with open("output.csv", "a", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(new_row)

实战 10：异常处理（健壮 CSV 读写工具）

python

运行

import csv def safe_read_csv(file_path): try: # 尝试正常读取 with open(file_path, "r", encoding="utf-8", newline="") as f: reader = csv.DictReader(f) return list(reader) # 捕获文件不存在异常 except FileNotFoundError: print("文件不存在") return [] # 捕获其他所有异常 except Exception as e: print(f"读取失败：{e}") return [] # 安全调用 data = safe_read_csv("data.csv")

实战 11：Excel 导出的制表符 CSV 读取

python

运行

import csv # Excel导出的制表符分隔文件，使用内置excel_tab方言 with open("excel_tab.csv", "r", encoding="utf-8") as f: reader = csv.reader(f, dialect=csv.excel_tab) for row in reader: print(row)

实战 12：空值 / 缺失字段自动填充

python

运行

import csv with open("data.csv", "r", encoding="utf-8") as f: reader = csv.DictReader(f) rows = [] for row in reader: # 年龄为空时自动填充为0 row["年龄"] = row["年龄"] if row["年龄"] else 0 rows.append(row) # 写入填充后的数据 with open("filled.csv", "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=reader.fieldnames) writer.writeheader() writer.writerows(rows)

实战 13：生成带序号的 CSV 报表

python

运行

import csv # 原始数据 data = [["张三", 23], ["李四", 25]] rows = [] # 生成序号从1开始 for i, item in enumerate(data, 1): rows.append([i, *item]) # 写入文件 header = ["序号", "姓名", "年龄"] with open("report.csv", "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(header) writer.writerows(rows)

实战 14：按条件修改 CSV 数据

python

运行

import csv with open("data.csv", "r", encoding="utf-8") as f: reader = csv.DictReader(f) rows = [] for row in reader: # 满足条件则修改字段 if row["城市"] == "北京": row["城市"] = "北京市" rows.append(row) # 写入修改后的数据 with open("updated.csv", "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=reader.fieldnames) writer.writeheader() writer.writerows(rows)

实战 15：数据库查询结果导出 CSV（MySQL）

python

运行

import csv import pymysql # 连接MySQL数据库 conn = pymysql.connect( host="localhost", user="root", password="123456", database="test" ) cursor = conn.cursor() # 执行查询SQL cursor.execute("SELECT name, age, city FROM user") rows = cursor.fetchall() # 写入CSV with open("db_export.csv", "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) # 写入表头 writer.writerow(["姓名", "年龄", "城市"]) # 写入数据库数据 writer.writerows(rows) # 关闭连接 cursor.close() conn.close()