第25到35篇学了十样本领：读CSV、读Excel、筛选排序、清洗、合并、分组、导出……可它们是”散装”的，每次都要现想现写。今天把它们串成一条流水线——原料（原始数据）从这头进去，成品（Excel报表）从那头出来，以后每月的报表活儿就是”换数据、按一次回车”。今天只引入一个新概念：把每一步写成函数、按顺序串起来，中间用”行数核对”当质检员
一、阶段3的”毕业答辩”
恭喜，你走到了阶段3（文件和数据）的最后一篇。先盘点一下军火库里有什么：
- 第25篇：
pd.read_csv()——把CSV文件整张吞进程序 - 第31篇：
pd.read_excel()——Excel也照吞不误 - 第32篇：筛选、排序——”把这几行挑出来”
- 第33篇：
groupby——按部门/按班级分组报数 - 第34篇：
merge——两张表对上号拼成一张 - 第35篇：
to_excel/ExcelWriter——装订成册交出去
单看每一招都不难，可现实世界的活儿从来不是”只用一招”。回忆一下这一个月的剧情：老板先问总量（31），再问明细（32），再问分组对比（33），再让你把订单表和员工表拼起来（34），最后让你导出Excel（35）。每一篇是一道独立的题，而真实工作是十道题连成一套卷。
更扎心的是：这套”卷子”每个月都要重考一次。这个月你东拼西凑写了五十行代码才交出报表，下个月数据一换，你又得从头再来一遍？那不叫学会了，那叫”每次都现学”。
所以毕业答辩的题目是：把整套流程写成一个程序，以后换数据重跑一次就出货。
这在数据行当里有个正经名字：数据分析流水线（pipeline，”管道”）——原始数据从管道这头灌进去，经过一道道工序（清洗、合并、分析），成品报表从那头流出来。工厂里装配线的每个工位只干一件事，你的程序也一样：每个函数是一道工序，数据顺着传送带走完全程。
二、今天的新概念：把步骤写成函数，串成一条线
“流水线”听起来高级，拆开看就是你已经会的两样东西的组合：函数（第7篇）+ 按顺序调用（第3篇就会了）。唯一的新思维是：给每个步骤定好”进什么、出什么”，然后把它们首尾相接。
为什么非得用函数？三个理由，一个比一个实在：
- 能单跑。清洗那步结果不对？单独跑
清洗()调试，不用从头跑全流程 - 能复用。下个月报表改个分组维度，只动分析那一步，别的工序原封不动
- 能检查。每道工序的产出都是一个明确的变量，跑完打印一眼就知道哪一环出了问题
先造”原料”，让今天每个代码块都能真跑。把下面存成 造数据.py 运行一次，生成两个原始文件：
import pandas as pd
# 原料1：订单表（故意埋了几个坑——空行、带"元"字的销售额）
订单 = pd.DataFrame({
"订单号": ["A001", "A002", "A003", "A004", "A005", "A006", None],
"姓名": ["小李", "小王", "小张", "小赵", "小王", "小李", None],
"销售额": ["8000元", "12000", "15300", "9500", "11000", "7000", None],
"日期": ["2026-09-01", "2026-09-03", "2026-09-10", "2026-09-15", "2026-09-20", "2026-09-28", None],
})
订单.to_csv("订单.csv", index=False, encoding="utf-8-sig")
# 原料2：员工表（部门花名册）
员工 = pd.DataFrame({
"姓名": ["小李", "小王", "小张", "小赵"],
"部门": ["华南", "华东", "华东", "华北"],
"入职年份": [2022, 2023, 2021, 2024],
})
员工.to_csv("员工.csv", index=False, encoding="utf-8-sig")
print("原料已备好：订单.csv、员工.csv")然后是今天的主角，完整的流水线程序 数据分析流水线.py。先整体看一眼长相，别急着逐行抠——五道工序，一个总调度：
import pandas as pd
# ========== 工序1：读数据 ==========
def 读数据():
订单 = pd.read_csv("订单.csv", encoding="utf-8-sig")
员工 = pd.read_csv("员工.csv", encoding="utf-8-sig")
print(f"[读数据] 订单{len(订单)}行，员工{len(员工)}行")
return 订单, 员工
# ========== 工序2：清洗 ==========
def 清洗(订单):
订单 = 订单.dropna(subset=["订单号"]) # 干掉整行空白的记录
订单["销售额"] = 订单["销售额"].str.replace("元", "").astype(int) # 去掉"元"，转成数字
print(f"[清洗] 剩{len(订单)}行（去掉了空行，销售额已转数字）")
return 订单
# ========== 工序3：合并 ==========
def 合并(订单, 员工):
宽表 = pd.merge(订单, 员工, on="姓名", how="left")
没配上 = 宽表["部门"].isna().sum()
print(f"[合并] 合并后{len(宽表)}行，没配上部门的{没配上}行")
return 宽表
# ========== 工序4：分析 ==========
def 分析(宽表):
汇总 = 宽表.groupby("部门").agg(
总销售额=("销售额", "sum"),
订单数=("销售额", "count"),
).reset_index().sort_values("总销售额", ascending=False)
print(f"[分析] 分组结果{len(汇总)}个部门")
return 汇总
# ========== 工序5：导出 ==========
def 导出(汇总, 宽表):
with pd.ExcelWriter("月度销售报表.xlsx") as 装订机:
汇总.to_excel(装订机, sheet_name="部门汇总", index=False)
宽表.to_excel(装订机, sheet_name="订单明细", index=False)
print("[导出] 月度销售报表.xlsx 已生成")
# ========== 总调度：一键跑全流程 ==========
def main():
print("===== 流水线启动 =====")
订单, 员工 = 读数据()
订单 = 清洗(订单)
宽表 = 合并(订单, 员工)
汇总 = 分析(宽表)
导出(汇总, 宽表)
print("===== 流水线完成，去发报表吧 =====")
if __name__ == "__main__":
main()运行输出长这样，每一行都是一道工序的”质检回执”：
===== 流水线启动 =====
[读数据] 订单7行，员工4行
[清洗] 剩6行（去掉了空行，销售额已转数字）
[合并] 合并后6行，没配上部门的0行
[分析] 分组结果3个部门
[导出] 月度销售报表.xlsx 已生成
===== 流水线完成，去发报表吧 =====打开 月度销售报表.xlsx：第一页是按部门汇总（华东 26300、华南 15000、华北 9500），第二页是带部门的订单明细。从两个脏兮兮的原始文件到能发老板的报表，一个回车。
先解释两个可能让你眨眼的写法：
if __name__ == "__main__": main()——这是Python程序的”标准点火开关”，意思是”直接运行这个文件时，才执行main()“。把它当成固定的”开机仪式”照抄就行，好处是以后别的程序想借用你的工序（import 数据分析流水线）时，不会一导入就自动跑全流程汇总 = 宽表.groupby("部门").agg(总销售额=("销售额", "sum"), ...)——第33篇groupby的进阶写法，agg一次问多个问题，还能顺便给结果列起人话名字（”总销售额”而不是”销售额_sum”）
三、流水线的骨架：进什么、出什么，先定接口
再看一遍五道工序的”长相”，你会发现一个规律——每个函数都是一进一出，首尾相接：
订单.csv ─┐
├→ 读数据() → 清洗() ──┐
员工.csv ─┘ ├→ 合并() → 分析() → 导出() → 月度销售报表.xlsx
┘读数据()不吃参数，吐出两个原料表清洗(订单)吃订单表，吐干净的订单表合并(订单, 员工)吃两张表，吐拼好的宽表分析(宽表)吃宽表，吐汇总表导出(汇总, 宽表)吃两张结果表，吐文件
函数的参数就是”传送带的入口”，return就是”传送带的出口”。上一道工序的return，接上下一道工序的参数——数据就顺着传送带流动起来了。这就是”流水线”三个字的全部秘密：没有黑魔法，就是接口清晰的函数按顺序排队。
那 print 那几行质检回执呢？那是流水线上的质检工位。每个”[]”开头的输出都在回答同一个问题：”这一站干完活，东西还对吗？”尤其是行数核对：读进来7行、清洗后6行（干掉1行空行，合理）、合并后还是6行（没多没少，合理）——数字对得上，才敢放心把半成品传给下一站。要是清洗后突然只剩1行，你立刻知道清洗规则写过头了，而不是等到最后打开报表才发现”数据怎么没了”。
四、一键跑 vs. 单步跑：流水线的两种用法
流水线写好后，日常用法就是”一键跑”：
if __name__ == "__main__":
main()换新数据？把新的 订单.csv、员工.csv 拷进文件夹，重跑一次，报表自动更新。以前手工做一小时的月度报表，现在是一分钟——而且第二个月、第十二个月都是一分钟。这就是”写下来的流程”值钱的地方：成本一次性付清，收益每月到账。
但一键跑有个前提：流水线本身已经调通。调试期的正确姿势是单步跑——不跑main()，在交互环境或者临时脚本里只跑一站：
import pandas as pd
# 只调"清洗"这一站：直接喂一张表进去，看出来的表对不对
订单 = pd.read_csv("订单.csv", encoding="utf-8-sig")
干净 = 清洗(订单)
print(干净.head()) # 亲眼看看清洗结果
print(干净.dtypes) # 确认销售额是不是真的变成了数字哪一站的输出不对，就单独跑哪一站、改哪一站——这比”跑全流程→看最终报表不对→猜是哪一站干的→再跑全流程”快十倍。流水线的调试原则：单站修理，全线贯通。
还有个小技巧：调试期可以在可疑的工序后面加一句 半成品.to_csv("_中间产物.csv", index=False)，把半成品倒出来用Excel亲眼看看。调通之后删掉这些临时出口——流水线的正式版只留一个出口：成品报表。
五、四个必踩的坑（先打疫苗）
坑1：merge报 KeyError，或者合并后行数暴涨。 报KeyError: '姓名'，九成是两边的列名不一致——一边叫”姓名”，另一边叫”员工姓名”或者末尾带了个空格。合并后行数暴涨（10行变30行），是on的那列有重复值，一配多，每行被”复印”了好几份。疫苗：合并前先print(订单.columns.tolist(), 员工.columns.tolist())核对列名；合并后立即核对行数——合并()函数里那句print(f"合并后{len(宽表)}行")就是干这个的。另外how="left"保住左表每一行，配不上的部门留空（NaN），这正是你想要的”不丢订单”。
坑2：sum()出来是个离谱的大数，或者报can only concatenate str。 比如”销售额”里混了一个”8000元”或者”暂无”，整列就变成了文字（字符串）——sum把它们首尾相接连成一条长串，或者直接报错。疫苗：清洗工序里”转数字”是必做动作——订单["销售额"] = 订单["销售额"].str.replace("元", "").astype(int)（去杂质→转整数）。转之前先print(宽表.dtypes)看一眼类型，object就说明还是文字。真有转不动的脏值（比如”暂无”），用 pd.to_numeric(订单["销售额"], errors="coerce") 把它变成空值，再决定删行还是补零——第29篇”数据清洗”的功夫就是为这一天准备的。
坑3：FileNotFoundError——”文件明明在啊！”。 程序说找不到订单.csv，你明明放了。九成是放错了地方：Python找文件是从”程序运行时所在的文件夹”找，不是从”py文件躺着的文件夹”找（两者常常不一致）。疫苗：原料文件和流水线程序放同一个文件夹，从那个文件夹里运行程序。还是找不到，就打印一下当前位置对质：print(Path.cwd())（cwd = current working directory，”现在站哪儿”）。老手的做法是把所有路径写成程序开头的变量（原料文件夹 = "."），一站改全改。
坑4：流水线跑完了，但打开的是上个月的旧报表。 程序没报错、打印”已生成”，可打开Excel怎么还是上个月的数据？两种可能：一是输出文件正在Excel里开着，Windows下Excel锁住文件，pandas写不进去却又没让你看到明显报错（35篇坑3的老朋友）；二是报表路径写错了，成品流到了别的文件夹，你看的还是旧文件。疫苗：跑之前关掉所有打开的报表；跑完看打印里写的路径和文件修改时间（Excel里”文件→信息”能看到修改时间）。流水线的质检回执说”已生成”，你还要亲眼确认”生成的是新的”——程序的完成≠活儿的完成。
六、跟AI协作的正确姿势（流水线特供）
- 描述需求时把整条线画给AI。别一句一句要代码，一次性说全：”写一条数据分析流水线：读取’订单.csv’和’员工.csv’→清洗（去空行、销售额去’元’转数字）→按姓名left合并→按部门汇总总销售额和订单数→导出’月度销售报表.xlsx’两个sheet（部门汇总、订单明细）。每步一个函数，每步打印行数核对，最后main()一键跑全流程“。工序、顺序、每站的进出、质检要求，四要素说全——AI直接给你整条能跑的线，而不是五段互相不认识的代码
- 调试时贴”工序回执”而不是只贴报错。流水线出问题，把从
[读数据]开始的全部打印和报错一起贴给AI——AI看一眼”清洗后剩1行”就知道问题出在清洗工序，你只贴最后一行报错它只能瞎猜
- 让AI”只改一站，别动其他”。加需求时明说：”只改分析()这一步，加一个’平均单值’列，其他工序保持不动“——流水线最怕AI好心帮你重构全部代码，五个调好的工序一夜回到解放前。拿到改动后，跑一遍单站+一遍全线，确认没牵连
高频句式：”把’读取→清洗→合并→分析→导出’写成一条数据分析流水线，每步一个函数、打印行数核对，main()一键跑，报错时能单跑某一步调试“。
七、今日小练习
练习1：给流水线加个体检站
在”读数据”和”清洗”之间插一道”体检”工序：打印每列有几个空值、销售额的最值（min/max）。（提示：新函数def 体检(表):，里面用表.isna().sum()看每列空值数、表["销售额"].min()看最小值。记得把体检站接进main()的传送带——参数接上一站的return。体检只看不改，return原表传给下一站。）
练习2：换条流水线——班级成绩报表
把原料换成”成绩.csv”（姓名、班级、语文、数学、英语），走同样的流程：读→清洗→按班级分组算平均分→导出xlsx。（提示：五道工序的骨架原样照抄，只改每站内部的列名和分组键——这正是函数化的好处：换数据不换流程。分组站把groupby("部门")换成groupby("班级")，agg问”平均分”用("语文", "mean")。）
练习3：流水线”半自动挡”
把文件名和分组键挪进配置文件（第28篇回锅）：config.json里写{"订单文件": "订单.csv", "员工文件": "员工.csv", "分组键": "部门", "输出文件": "月度销售报表.xlsx"}，流水线从配置文件读参数。（提示：开头import json，配置 = json.load(open("config.json", encoding="utf-8"))，然后把程序里写死的字符串全换成配置["订单文件"]这样的写法。改配置不动代码——下个月老板改需求，你只编辑config.json。）
写在后面：从”会十招”到”一套组合拳”
盘点今天：流水线=函数化的工序按顺序排队——读数据→清洗→合并→分析→导出，每站”进什么、出什么”接口清晰，每站打印行数当质检，main()一键开动，出问题单站修理。今天引入的新概念其实只有一句话：把步骤写成函数、定好接口串起来——但正是这一句话，让你从”会十招散手”变成”一套完整组合拳”。
更重要的是视角变了：以前你是”被问题追着跑”，老板问一句你写一段；现在你是”守着一条生产线”，数据进来、报表出去。会做一次性分析的人到处都是，能把分析写成流水线、每月自动出货的人才是团队里那个”报表一小时出、还从不出错”的狠角色。
至此，阶段3（文件和数据）正式毕业。回望这12篇：从CSV这个”白开水格式”起步，到pandas把整张表一口吞下，再到清洗、合并、分组、导出，最后串成流水线——数据这关，你过了。
下一篇，开一个全新的大台阶：阶段4（网络基础）。你的程序不再只跟本地文件打交道，而是要走出电脑，去网络上取数据——第一课，认识互联网的”点餐语言”HTTP。
下篇预告
第三十七篇：HTTP请求入门 — 让程序上网”点菜”——网页、APP、AI，全都在用一种叫HTTP的”点餐语言”跟服务器对话。requests.get()一行，你的程序也能上网下单取数据——从”只看自家文件”到”能看全世界”，就从这一篇开始。
发表回复