080、综合案例从数据清洗到可视化报告的完整 Python 数据分析流水线一个让我熬夜到凌晨三点的坑上周帮业务部门处理一份销售数据客户说“数据很干净直接跑分析就行”。我信了结果一跑脚本报错信息像瀑布一样刷屏。最离谱的是某列日期字段里混着“2023-02-30”这种不存在的日期还有一行数据里金额字段写着“一万两千元”——中文大写。这种“干净”数据差点让我把键盘砸了。从那以后我养成了一个习惯不管谁跟我说数据干净先跑一遍清洗流水线再说。今天这篇笔记就是那次踩坑后总结的一套完整方案从原始数据到可视化报告一条龙搞定。数据流水线的骨架设计先搭框架再填肉。我习惯把整个流程拆成四个阶段数据加载与探查、清洗与转换、分析与聚合、可视化输出。每个阶段独立成函数方便调试时单独跑。importpandasaspdimportnumpyasnpimportmatplotlib.pyplotaspltimportseabornassnsfromdatetimeimportdatetimeimportwarnings warnings.filterwarnings(ignore)# 别学我这样粗暴生产环境要谨慎# 设置中文字体不然图表里全是方框plt.rcParams[font.sans-serif][SimHei]plt.rcParams[axes.unicode_minus]False这里踩过坑matplotlib默认字体不支持中文不设置的话图表标题和标签全是□□□□。SimHei是黑体大部分Windows系统自带Mac用户换成’PingFang SC’。第一阶段数据加载与探查拿到数据先别急着分析用info()和describe()摸个底。我习惯写个探查函数把常见问题一次性暴露出来。defload_and_explore(file_path): 加载数据并做初步探查 别这样写直接pd.read_csv()就开干 try:dfpd.read_csv(file_path,encodingutf-8)exceptUnicodeDecodeError:# 这里踩过坑有些CSV文件是GBK编码直接读会崩dfpd.read_csv(file_path,encodinggbk)print(*50)print(数据概览)print(f行数{df.shape[0]}, 列数{df.shape[1]})print(\n列名与数据类型)print(df.dtypes)print(\n缺失值统计)print(df.isnull().sum())print(\n前5行数据)print(df.head())returndf这个函数看起来简单但那个try-except编码处理救过我很多次。业务部门导出的CSV文件编码格式五花八门UTF-8和GBK是最常见的两种。第二阶段数据清洗——最磨人的环节清洗函数我写得特别啰嗦因为每个判断都可能藏着坑。下面这个函数处理了最常见的三类脏数据缺失值、异常值、格式问题。defclean_data(df): 数据清洗主函数 注意这里会修改原始DataFrame建议先copy() df_cleandf.copy()# 保留原始数据方便回溯# 1. 处理缺失值# 别这样写直接df.dropna()会删掉太多有效数据forcolindf_clean.columns:ifdf_clean[col].dtypeobject:# 字符串列用众数填充这里踩过坑用Unknown填充会引入新问题mode_valdf_clean[col].mode()ifnotmode_val.empty:df_clean[col].fillna(mode_val[0],inplaceTrue)else:# 数值列用中位数填充比均值更抗异常值df_clean[col].fillna(df_clean[col].median(),inplaceTrue)# 2. 处理重复行dup_countdf_clean.duplicated().sum()ifdup_count0:print(f发现{dup_count}行重复数据已删除)df_clean.drop_duplicates(inplaceTrue)# 3. 处理日期字段——这里踩过大坑if日期indf_clean.columns:# 先尝试标准格式不行再逐个处理df_clean[日期]pd.to_datetime(df_clean[日期],errorscoerce)# 检查转换失败的日期bad_datesdf_clean[日期].isnull().sum()ifbad_dates0:print(f警告{bad_dates}个日期无法解析已转为NaT)# 这里可以加更复杂的日期解析逻辑比如处理2023年1月5日这种格式# 4. 处理金额字段——中文大写转数字if金额indf_clean.columns:# 检查是否有非数字内容non_numericdf_clean[金额].apply(lambdax:notstr(x).replace(.,).isdigit())ifnon_numeric.any():print(发现非数字金额尝试转换...)df_clean[金额]df_clean[金额].apply(convert_chinese_amount)returndf_cleandefconvert_chinese_amount(val): 中文大写金额转数字 别这样写用一堆if-else维护起来想哭 # 这里只处理简单情况完整版需要更复杂的映射mapping{零:0,壹:1,贰:2,叁:3,肆:4,伍:5,陆:6,柒:7,捌:8,玖:9,拾:10,佰:100,仟:1000,万:10000,元:,整:}try:val_strstr(val)fork,vinmapping.items():val_strval_str.replace(k,v)# 简单处理如果是纯数字就返回否则返回NaNifval_str.replace(.,).isdigit():returnfloat(val_str)else:returnnp.nanexcept:returnnp.nan这个清洗函数看起来长但每个步骤都是血泪教训换来的。特别是日期处理那块pd.to_datetime的errorscoerce’参数是神器不会因为一条坏数据就让整个程序崩溃。第三阶段数据分析与聚合清洗完数据终于可以开始正经分析了。我习惯先做几个基础统计再根据业务需求做分组聚合。defanalyze_data(df): 数据分析核心逻辑 这里踩过坑直接对全表做统计内存会爆 # 1. 基础统计print(\n数值列统计描述)print(df.describe())# 2. 分组分析——按月份和产品类别if日期indf.columnsand类别indf.columns:# 提取月份df[月份]df[日期].dt.month# 别这样写用for循环逐行计算效率极低monthly_salesdf.groupby([月份,类别])[金额].agg([sum,mean,count])monthly_sales.columns[总销售额,平均销售额,订单数]monthly_salesmonthly_sales.reset_index()print(\n月度销售统计按类别)print(monthly_sales.head(20))returnmonthly_salesreturnNonegroupbyagg的组合拳比手动循环快几十倍。这里有个小技巧agg可以传多个聚合函数一次性算出sum、mean、count省得写多个groupby。第四阶段可视化报告生成分析结果不能只留在终端里得生成能拿给老板看的图表。我习惯用matplotlibseaborn组合虽然不如plotly交互性强但胜在稳定、不依赖网络。defgenerate_report(monthly_data,output_pathreport.png): 生成可视化报告 别这样写把所有图表塞进一个figure会乱成一团 fig,axesplt.subplots(2,2,figsize(16,12))# 图1月度总销售额趋势ax1axes[0,0]monthly_totalmonthly_data.groupby(月份)[总销售额].sum()ax1.plot(monthly_total.index,monthly_total.values,markero,linewidth2,color#2E86AB)ax1.set_title(月度总销售额趋势,fontsize14,fontweightbold)ax1.set_xlabel(月份)ax1.set_ylabel(销售额元)ax1.grid(True,alpha0.3)# 图2各类别销售额占比——这里踩过坑标签重叠ax2axes[0,1]category_totalmonthly_data.groupby(类别)[总销售额].sum()# 别这样写直接用默认参数小类别标签会挤在一起wedges,texts,autotextsax2.pie(category_total.values,labelsNone,# 先不显示标签autopct%1.1f%%,startangle90,colorssns.color_palette(husl,len(category_total)))# 用图例代替直接标签更清晰ax2.legend(wedges,category_total.index,title类别,loccenter left,bbox_to_anchor(1,0,0.5,1))ax2.set_title(各类别销售额占比,fontsize14,fontweightbold)# 图3平均销售额对比柱状图ax3axes[1,0]avg_salesmonthly_data.groupby(类别)[平均销售额].mean().sort_values()barsax3.barh(avg_sales.index,avg_sales.values,colorsns.color_palette(viridis,len(avg_sales)))ax3.set_title(各类别平均销售额对比,fontsize14,fontweightbold)ax3.set_xlabel(平均销售额元)# 在柱子上显示数值forbar,valinzip(bars,avg_sales.values):ax3.text(val10,bar.get_y()bar.get_height()/2,f{val:.0f},vacenter)# 图4订单数分布热力图ax4axes[1,1]pivot_datamonthly_data.pivot_table(values订单数,index月份,columns类别,aggfuncsum)sns.heatmap(pivot_data,annotTrue,fmt.0f,cmapYlOrRd,axax4)ax4.set_title(月度订单数分布热力图,fontsize14,fontweightbold)ax4.set_xlabel(类别)ax4.set_ylabel(月份)plt.tight_layout()plt.savefig(output_path,dpi150,bbox_inchestight)print(f报告已保存至{output_path})plt.show()这个报告函数我调试了最久。特别是饼图的标签问题默认情况下小类别的标签会叠在一起根本看不清。改用图例后清爽多了。热力图的annot参数可以显示数值但要注意fmt格式不然小数点后一堆零。组装流水线最后把四个阶段串起来形成一个完整的流水线函数。defdata_pipeline(file_path,output_reportsales_report.png): 完整的数据分析流水线 别这样写把所有代码塞进一个函数调试时想哭 print(*60)print(开始执行数据分析流水线)print(*60)# 阶段1加载与探查print(\n【阶段1】数据加载与探查)dfload_and_explore(file_path)# 阶段2数据清洗print(\n【阶段2】数据清洗)df_cleanclean_data(df)print(f清洗后数据量{df_clean.shape[0]}行 x{df_clean.shape[1]}列)# 阶段3数据分析print(\n【阶段3】数据分析)monthly_dataanalyze_data(df_clean)# 阶段4可视化报告print(\n【阶段4】生成可视化报告)generate_report(monthly_data,output_report)print(\n*60)print(流水线执行完毕)print(*60)returndf_clean,monthly_data# 执行流水线if__name____main__:# 这里踩过坑路径用相对路径容易找不到文件clean_data,analysis_resultdata_pipeline(file_path./data/sales_data.csv,output_report./output/sales_analysis_report.png)个人经验性建议写这个流水线的时候我反复改了好几版。有些经验分享给你永远不要相信原始数据是干净的。哪怕对方拍胸脯保证也要跑一遍清洗流程。我吃过太多亏了最离谱的一次是某列数据里混着全角半角字符肉眼根本看不出来。清洗函数要设计成可重入的。也就是说跑两次清洗不会出问题。有些新手写的清洗代码第一次跑把空值填了第二次跑又把填好的值当成异常处理了。可视化报告要留白。别把所有图表塞进一张图里老板看不懂你也解释不清。我一般控制在4-6个图表每个图表只讲一个故事。调试时用print上线后用logging。开发阶段print大法好但生产环境一定要换成logging不然日志文件会爆炸。最后一条也是最重要的写代码前先想清楚业务逻辑。我见过太多人上来就写groupby结果分组维度错了分析结果完全跑偏。先跟业务方确认清楚你要看什么为什么看这个怎么看才有意义这套流水线我用了大半年从销售数据到用户行为分析改改参数就能复用。下次遇到“很干净”的数据记得先跑一遍清洗——相信我你不会后悔的。
080、综合案例：从数据清洗到可视化报告的完整 Python 数据分析流水线
080、综合案例从数据清洗到可视化报告的完整 Python 数据分析流水线一个让我熬夜到凌晨三点的坑上周帮业务部门处理一份销售数据客户说“数据很干净直接跑分析就行”。我信了结果一跑脚本报错信息像瀑布一样刷屏。最离谱的是某列日期字段里混着“2023-02-30”这种不存在的日期还有一行数据里金额字段写着“一万两千元”——中文大写。这种“干净”数据差点让我把键盘砸了。从那以后我养成了一个习惯不管谁跟我说数据干净先跑一遍清洗流水线再说。今天这篇笔记就是那次踩坑后总结的一套完整方案从原始数据到可视化报告一条龙搞定。数据流水线的骨架设计先搭框架再填肉。我习惯把整个流程拆成四个阶段数据加载与探查、清洗与转换、分析与聚合、可视化输出。每个阶段独立成函数方便调试时单独跑。importpandasaspdimportnumpyasnpimportmatplotlib.pyplotaspltimportseabornassnsfromdatetimeimportdatetimeimportwarnings warnings.filterwarnings(ignore)# 别学我这样粗暴生产环境要谨慎# 设置中文字体不然图表里全是方框plt.rcParams[font.sans-serif][SimHei]plt.rcParams[axes.unicode_minus]False这里踩过坑matplotlib默认字体不支持中文不设置的话图表标题和标签全是□□□□。SimHei是黑体大部分Windows系统自带Mac用户换成’PingFang SC’。第一阶段数据加载与探查拿到数据先别急着分析用info()和describe()摸个底。我习惯写个探查函数把常见问题一次性暴露出来。defload_and_explore(file_path): 加载数据并做初步探查 别这样写直接pd.read_csv()就开干 try:dfpd.read_csv(file_path,encodingutf-8)exceptUnicodeDecodeError:# 这里踩过坑有些CSV文件是GBK编码直接读会崩dfpd.read_csv(file_path,encodinggbk)print(*50)print(数据概览)print(f行数{df.shape[0]}, 列数{df.shape[1]})print(\n列名与数据类型)print(df.dtypes)print(\n缺失值统计)print(df.isnull().sum())print(\n前5行数据)print(df.head())returndf这个函数看起来简单但那个try-except编码处理救过我很多次。业务部门导出的CSV文件编码格式五花八门UTF-8和GBK是最常见的两种。第二阶段数据清洗——最磨人的环节清洗函数我写得特别啰嗦因为每个判断都可能藏着坑。下面这个函数处理了最常见的三类脏数据缺失值、异常值、格式问题。defclean_data(df): 数据清洗主函数 注意这里会修改原始DataFrame建议先copy() df_cleandf.copy()# 保留原始数据方便回溯# 1. 处理缺失值# 别这样写直接df.dropna()会删掉太多有效数据forcolindf_clean.columns:ifdf_clean[col].dtypeobject:# 字符串列用众数填充这里踩过坑用Unknown填充会引入新问题mode_valdf_clean[col].mode()ifnotmode_val.empty:df_clean[col].fillna(mode_val[0],inplaceTrue)else:# 数值列用中位数填充比均值更抗异常值df_clean[col].fillna(df_clean[col].median(),inplaceTrue)# 2. 处理重复行dup_countdf_clean.duplicated().sum()ifdup_count0:print(f发现{dup_count}行重复数据已删除)df_clean.drop_duplicates(inplaceTrue)# 3. 处理日期字段——这里踩过大坑if日期indf_clean.columns:# 先尝试标准格式不行再逐个处理df_clean[日期]pd.to_datetime(df_clean[日期],errorscoerce)# 检查转换失败的日期bad_datesdf_clean[日期].isnull().sum()ifbad_dates0:print(f警告{bad_dates}个日期无法解析已转为NaT)# 这里可以加更复杂的日期解析逻辑比如处理2023年1月5日这种格式# 4. 处理金额字段——中文大写转数字if金额indf_clean.columns:# 检查是否有非数字内容non_numericdf_clean[金额].apply(lambdax:notstr(x).replace(.,).isdigit())ifnon_numeric.any():print(发现非数字金额尝试转换...)df_clean[金额]df_clean[金额].apply(convert_chinese_amount)returndf_cleandefconvert_chinese_amount(val): 中文大写金额转数字 别这样写用一堆if-else维护起来想哭 # 这里只处理简单情况完整版需要更复杂的映射mapping{零:0,壹:1,贰:2,叁:3,肆:4,伍:5,陆:6,柒:7,捌:8,玖:9,拾:10,佰:100,仟:1000,万:10000,元:,整:}try:val_strstr(val)fork,vinmapping.items():val_strval_str.replace(k,v)# 简单处理如果是纯数字就返回否则返回NaNifval_str.replace(.,).isdigit():returnfloat(val_str)else:returnnp.nanexcept:returnnp.nan这个清洗函数看起来长但每个步骤都是血泪教训换来的。特别是日期处理那块pd.to_datetime的errorscoerce’参数是神器不会因为一条坏数据就让整个程序崩溃。第三阶段数据分析与聚合清洗完数据终于可以开始正经分析了。我习惯先做几个基础统计再根据业务需求做分组聚合。defanalyze_data(df): 数据分析核心逻辑 这里踩过坑直接对全表做统计内存会爆 # 1. 基础统计print(\n数值列统计描述)print(df.describe())# 2. 分组分析——按月份和产品类别if日期indf.columnsand类别indf.columns:# 提取月份df[月份]df[日期].dt.month# 别这样写用for循环逐行计算效率极低monthly_salesdf.groupby([月份,类别])[金额].agg([sum,mean,count])monthly_sales.columns[总销售额,平均销售额,订单数]monthly_salesmonthly_sales.reset_index()print(\n月度销售统计按类别)print(monthly_sales.head(20))returnmonthly_salesreturnNonegroupbyagg的组合拳比手动循环快几十倍。这里有个小技巧agg可以传多个聚合函数一次性算出sum、mean、count省得写多个groupby。第四阶段可视化报告生成分析结果不能只留在终端里得生成能拿给老板看的图表。我习惯用matplotlibseaborn组合虽然不如plotly交互性强但胜在稳定、不依赖网络。defgenerate_report(monthly_data,output_pathreport.png): 生成可视化报告 别这样写把所有图表塞进一个figure会乱成一团 fig,axesplt.subplots(2,2,figsize(16,12))# 图1月度总销售额趋势ax1axes[0,0]monthly_totalmonthly_data.groupby(月份)[总销售额].sum()ax1.plot(monthly_total.index,monthly_total.values,markero,linewidth2,color#2E86AB)ax1.set_title(月度总销售额趋势,fontsize14,fontweightbold)ax1.set_xlabel(月份)ax1.set_ylabel(销售额元)ax1.grid(True,alpha0.3)# 图2各类别销售额占比——这里踩过坑标签重叠ax2axes[0,1]category_totalmonthly_data.groupby(类别)[总销售额].sum()# 别这样写直接用默认参数小类别标签会挤在一起wedges,texts,autotextsax2.pie(category_total.values,labelsNone,# 先不显示标签autopct%1.1f%%,startangle90,colorssns.color_palette(husl,len(category_total)))# 用图例代替直接标签更清晰ax2.legend(wedges,category_total.index,title类别,loccenter left,bbox_to_anchor(1,0,0.5,1))ax2.set_title(各类别销售额占比,fontsize14,fontweightbold)# 图3平均销售额对比柱状图ax3axes[1,0]avg_salesmonthly_data.groupby(类别)[平均销售额].mean().sort_values()barsax3.barh(avg_sales.index,avg_sales.values,colorsns.color_palette(viridis,len(avg_sales)))ax3.set_title(各类别平均销售额对比,fontsize14,fontweightbold)ax3.set_xlabel(平均销售额元)# 在柱子上显示数值forbar,valinzip(bars,avg_sales.values):ax3.text(val10,bar.get_y()bar.get_height()/2,f{val:.0f},vacenter)# 图4订单数分布热力图ax4axes[1,1]pivot_datamonthly_data.pivot_table(values订单数,index月份,columns类别,aggfuncsum)sns.heatmap(pivot_data,annotTrue,fmt.0f,cmapYlOrRd,axax4)ax4.set_title(月度订单数分布热力图,fontsize14,fontweightbold)ax4.set_xlabel(类别)ax4.set_ylabel(月份)plt.tight_layout()plt.savefig(output_path,dpi150,bbox_inchestight)print(f报告已保存至{output_path})plt.show()这个报告函数我调试了最久。特别是饼图的标签问题默认情况下小类别的标签会叠在一起根本看不清。改用图例后清爽多了。热力图的annot参数可以显示数值但要注意fmt格式不然小数点后一堆零。组装流水线最后把四个阶段串起来形成一个完整的流水线函数。defdata_pipeline(file_path,output_reportsales_report.png): 完整的数据分析流水线 别这样写把所有代码塞进一个函数调试时想哭 print(*60)print(开始执行数据分析流水线)print(*60)# 阶段1加载与探查print(\n【阶段1】数据加载与探查)dfload_and_explore(file_path)# 阶段2数据清洗print(\n【阶段2】数据清洗)df_cleanclean_data(df)print(f清洗后数据量{df_clean.shape[0]}行 x{df_clean.shape[1]}列)# 阶段3数据分析print(\n【阶段3】数据分析)monthly_dataanalyze_data(df_clean)# 阶段4可视化报告print(\n【阶段4】生成可视化报告)generate_report(monthly_data,output_report)print(\n*60)print(流水线执行完毕)print(*60)returndf_clean,monthly_data# 执行流水线if__name____main__:# 这里踩过坑路径用相对路径容易找不到文件clean_data,analysis_resultdata_pipeline(file_path./data/sales_data.csv,output_report./output/sales_analysis_report.png)个人经验性建议写这个流水线的时候我反复改了好几版。有些经验分享给你永远不要相信原始数据是干净的。哪怕对方拍胸脯保证也要跑一遍清洗流程。我吃过太多亏了最离谱的一次是某列数据里混着全角半角字符肉眼根本看不出来。清洗函数要设计成可重入的。也就是说跑两次清洗不会出问题。有些新手写的清洗代码第一次跑把空值填了第二次跑又把填好的值当成异常处理了。可视化报告要留白。别把所有图表塞进一张图里老板看不懂你也解释不清。我一般控制在4-6个图表每个图表只讲一个故事。调试时用print上线后用logging。开发阶段print大法好但生产环境一定要换成logging不然日志文件会爆炸。最后一条也是最重要的写代码前先想清楚业务逻辑。我见过太多人上来就写groupby结果分组维度错了分析结果完全跑偏。先跟业务方确认清楚你要看什么为什么看这个怎么看才有意义这套流水线我用了大半年从销售数据到用户行为分析改改参数就能复用。下次遇到“很干净”的数据记得先跑一遍清洗——相信我你不会后悔的。
