第 0 步 · 为什么是 CSV
CSV(逗号分隔值)是表格数据的事实标准:Excel 能导、数据库能导、AI 训练数据常用。一行 = 一条记录,第一行往往是列名:
name,subject,score,date
丢丢,数学,92,2026-08-01
丢丢,英语,88,2026-08-02
Riemann,Python,95,2026-08-03
本节点学标准库 csv + statistics,不依赖第三方库——先搞懂"数据流水线"的骨架,pandas 那套高级武器以后在 AI 课程里学。
第 1 步 · 读 CSV(csv 模块)
import csv
# 方式 1:读成行(每行是列表)
with open("grades.csv", "r", encoding="utf-8") as f:
rows = list(csv.reader(f))
# rows[0] = ['name','subject','score','date'] 表头
# rows[1] = ['丢丢','数学','92','2026-08-01'] 数据行
# 方式 2:读成字典(表头当键,最好用)
with open("grades.csv", "r", encoding="utf-8") as f:
records = list(csv.DictReader(f))
# records[0] = {'name':'丢丢','subject':'数学','score':'92','date':'2026-08-01'}
# 注意:读出来全是字符串!score 要 int() 才能算
CSV 读出来全是 str——这是新手最大的惊讶。做统计前必须类型转换:
score = int(row["score"])。第 2 步 · 写 CSV
import csv
rows = [
["name", "subject", "score"],
["丢丢", "数学", 92],
["丢丢", "英语", 88],
]
with open("out.csv", "w", encoding="utf-8", newline="") as f:
csv.writer(f).writerows(rows)
# 字典写入(与 DictReader 对应)
with open("out.csv", "w", encoding="utf-8", newline="") as f:
w = csv.DictWriter(f, fieldnames=["name", "subject", "score"])
w.writeheader()
w.writerow({"name": "丢丢", "subject": "数学", "score": 92})
写文件必须 newline="":Windows 下不加会在行间多空行。这是 csv 写入的经典坑,养成习惯直接写上。
第 3 步 · statistics(统计三兄弟)
import statistics
scores = [92, 88, 95, 76, 83]
statistics.mean(scores) # 86.8 平均值
statistics.median(scores) # 88 中位数(不受极端值影响)
statistics.stdev(scores) # 标准差:波动有多大
# 最大最小(内置函数)
max(scores); min(scores); sum(scores) / len(scores) # 手写平均也行
mean vs median:团队平均薪资会被几个高薪拉高,中位数更能代表"大多数"。看数据先问:我该用哪个?
第 4 步 · 数据流水线(洗数据的标准四步)
真实数据永远是脏的。四个步骤缺一不可:
import csv, statistics
# ① 读取
with open("grades.csv", "r", encoding="utf-8") as f:
records = list(csv.DictReader(f))
# ② 清洗:转类型 + 丢脏行
clean = []
for r in records:
try:
score = int(r["score"]) # 转数字,失败说明这行脏
except ValueError:
continue # 跳过坏行(记录到日志更好)
if score < 0 or score > 100: # 逻辑校验
continue
clean.append((r["name"], r["subject"], score))
# ③ 统计
scores = [s for _, _, s in clean]
avg = statistics.mean(scores)
# ④ 输出(报告 / 写回文件 / 下一步)
print(f"有效记录 {len(clean)} 条,平均分 {avg:.1f}")
把"清洗"当成独立环节:真实项目 60% 的时间在洗数据。宁可多写校验,别让脏数据悄悄进统计——坏数据比没数据更危险(GIGO:垃圾进,垃圾出)。
第 5 步 · 实战项目:成绩分析报告
需求:读 grades.csv,输出每人平均分排名、全班总分平均、每科平均,并把报告写成 txt 存档。
import csv, statistics
from collections import defaultdict
from datetime import datetime
# 读取 + 清洗 + 分组
by_person = defaultdict(list)
by_subject = defaultdict(list)
with open("grades.csv", "r", encoding="utf-8") as f:
for r in csv.DictReader(f):
try:
s = int(r["score"])
except ValueError:
continue
by_person[r["name"]].append(s)
by_subject[r["subject"]].append(s)
# 统计 + 生成报告行
lines = [f"成绩分析报告 · {datetime.now().strftime('%Y-%m-%d %H:%M')}", "=" * 30]
# 每人平均分排名
ranking = sorted(by_person.items(), key=lambda x: statistics.mean(x[1]), reverse=True)
lines.append("【个人排名】")
for i, (name, scores) in enumerate(ranking, 1):
lines.append(f"{i}. {name}: {statistics.mean(scores):.1f}")
# 每科平均
lines.append(""); lines.append("【各科平均】")
for subject, scores in by_subject.items():
lines.append(f"{subject}: {statistics.mean(scores):.1f}")
# 存档
with open("report.txt", "w", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
print("\n".join(lines))
这就是"数据流水线"的最小完整版:读 → 洗 → 算 → 存。数据分析、报表、AI 数据预处理全是这个骨架。跑通它,pandas 只是把这个骨架加速。
第 6 步 · 预告:pandas 在 AI 课程等你
本节点用标准库演示了流水线。真实世界(尤其 AI 数据)用 pandas(DataFrame 表格对象):一行 df.groupby("subject")["score"].mean() 完成上面半个程序。但它的学习曲线需要先懂"读→洗→算→存"的心智模型——你已经有了。
第 7 步 · 报错手册(本节点新增)
1. 统计结果全错 —— 把字符串当数字算了('92' 按文本排序/拼接)。修法:确认清洗阶段
int() 了。2. statistics.StatisticsError —— 传了空列表(所有行都被清洗丢了)。修法:先
if not scores 判断再统计。3. csv 读取中文乱码 —— 文件不是 utf-8(Excel 存的可能是 gbk)。修法:读时换
encoding="gbk" 试,或用 utf-8-sig(带 BOM)。4. 写出的 CSV 在 Excel 打开乱码 —— 中文 + utf-8。修法:写时用
encoding="utf-8-sig"。节点自测清单
- 徒手写出:DictReader 读取 CSV 并打印每个名字
- 徒手写出:清洗环节(转 int + 丢坏行 + 逻辑校验)
- 能说清 mean / median 的区别与选用
- 徒手写出:统计全班总分平均和最高分
- 能独立完成成绩分析报告(排名 + 各科平均 + 存档)
- 知道 csv 写入为什么要 newline=""
全勾 → 节点 ⑦ 完成,进入节点 ⑧(网络与 API)。未勾 → 回看对应步骤。