CSV 看起来只是用逗号分隔的文本,但真实数据通常还会包含表头、引号、空值、重复记录和格式不一致的字段。掌握 Python 标准库中的 csv 模块,不仅能完成日常数据处理,也能为面试中的文件解析题建立一套稳定的思路。
这篇文章用一组循序渐进的练习串起 CSV 处理流程:先读取记录,再进行字段转换和过滤,最后完成分组统计。示例数据是为了演示而构造的,你可以替换成自己的 CSV 文件继续练习。
先明确解析边界
处理 CSV 时,不建议直接使用 line.split(",")。字段本身可能包含逗号,例如:
id,name,comment
1,Alice,"likes Python, SQL, and data analysis"
如果直接按逗号切分,comment 会被错误拆成多个字段。Python 的 csv 模块会处理引号、转义字符和不同的换行方式,因此更适合作为默认方案。
下面的练习文件可以直接保存为 orders.csv:
order_id,customer,amount,status
1001,Alice,120.50,paid
1002,Bob,89.00,cancelled
1003,Alice,45.50,paid
1004,Carol,,paid
1005,Bob,210.00,paid
练习一:读取并检查记录
使用 csv.DictReader 可以把每一行转换成字典。相比通过下标访问列,字典方式更容易读懂,也不容易因为列顺序变化而读错数据。
import csv
from pathlib import Path
def read_orders(path: str) -> list[dict[str, str]]:
with Path(path).open("r", encoding="utf-8", newline="") as file:
reader = csv.DictReader(file)
if reader.fieldnames != ["order_id", "customer", "amount", "status"]:
raise ValueError(f"unexpected columns: {reader.fieldnames}")
return list(reader)
if __name__ == "__main__":
orders = read_orders("orders.csv")
for order in orders:
print(order)
运行:
python parse_orders.py
这里有三个值得保留的习惯:使用 encoding="utf-8" 明确字符集,使用 newline="" 配合 csv 模块处理换行,并在输入边界检查表头。文件格式一旦变化,尽早失败通常比生成一份看似正常但实际错误的结果更容易排查。
练习二:转换类型并过滤数据
DictReader 返回的值都是字符串。金额需要转换成数值,缺失金额则必须明确处理。对于金额这类财务字段,生产代码通常应使用 decimal.Decimal,避免二进制浮点数带来的精度问题。
下面的程序只统计已支付且金额完整的订单:
import csv
from decimal import Decimal, InvalidOperation
from pathlib import Path
def paid_orders(path: str) -> list[dict[str, object]]:
result = []
with Path(path).open("r", encoding="utf-8", newline="") as file:
for line_number, row in enumerate(csv.DictReader(file), start=2):
raw_amount = (row.get("amount") or "").strip()
if not raw_amount:
print(f"skip line {line_number}: missing amount")
continue
try:
amount = Decimal(raw_amount)
except InvalidOperation:
print(f"skip line {line_number}: invalid amount {raw_amount!r}")
continue
if row.get("status", "").strip().lower() != "paid":
continue
result.append({
"order_id": row["order_id"].strip(),
"customer": row["customer"].strip(),
"amount": amount,
})
return result
if __name__ == "__main__":
for order in paid_orders("orders.csv"):
print(order["order_id"], order["customer"], order["amount"])
这道练习的重点不是语法,而是定义输入异常的行为:缺失金额是跳过、记为零,还是直接终止?不同业务有不同答案。面试中应把这个选择说出来,而不是让异常行为隐藏在类型转换代码里。
练习三:按客户汇总金额
读取、清洗和业务计算可以拆成独立步骤。这样既方便测试,也能在未来加入排序、输出新 CSV 或替换输入来源。
from collections import defaultdict
from decimal import Decimal
def total_by_customer(orders: list[dict[str, object]]) -> dict[str, Decimal]:
totals: dict[str, Decimal] = defaultdict(Decimal)
for order in orders:
customer = str(order["customer"])
amount = order["amount"]
if not isinstance(amount, Decimal):
raise TypeError("amount must be Decimal")
totals[customer] += amount
return dict(totals)
可以把前面的函数组合起来运行:
from parse_orders import paid_orders
from report import total_by_customer
orders = paid_orders("orders.csv")
for customer, total in sorted(total_by_customer(orders).items()):
print(f"{customer}: {total:.2f}")
对于示例数据,Alice 的已支付金额应为 166.00,Bob 的已支付金额应为 210.00。Carol 因为金额为空,会在清洗阶段被跳过。
面试中可以继续追问
完成基础实现后,可以给自己增加约束,让练习更接近实际面试:
- CSV 文件很大,不能一次性使用
list(reader),如何改成逐行处理? - 文件使用分号而不是逗号作为分隔符,应该修改哪里?
- 某个字段包含换行符或逗号时,当前实现是否仍然正确?
- 如何把汇总结果写入新的 CSV 文件?
- 如何报告所有坏行,而不是遇到第一条错误就停止?
- 如何为缺少表头、重复订单号和负金额增加校验?
大文件场景下,核心变化通常只是避免把所有记录放进内存:
import csv
def count_paid_rows(path: str) -> int:
count = 0
with open(path, "r", encoding="utf-8", newline="") as file:
for row in csv.DictReader(file):
if row.get("status", "").strip().lower() == "paid":
count += 1
return count
一份可执行的检查清单
练习 CSV 解析题时,可以按下面的顺序组织答案:
- 明确文件编码、分隔符和表头结构。
- 使用
csv.reader或csv.DictReader,不要手写逗号切分。 - 为必填字段做存在性和格式校验。
- 在边界处完成字符串到
int、Decimal等类型的转换。 - 明确坏数据的处理策略,并保留行号或错误信息。
- 让读取、清洗、计算和写出保持相对独立。
- 用空文件、缺失字段、带逗号的字段和非法金额补充测试。
CSV 解析题真正考查的是对输入边界的判断。只会读出一行数据不难;能在数据不完美、文件很大、字段规则变化时保持结果可解释,才是可以迁移到真实项目中的能力。