Python数据清洗实战：从zip文件到DataFrame的完整流程

简介数据清洗数据源.zip是一份面向大数据分析与数据工程初学者的实战数据包定位服务于数据清洗教学、课程实训与自学练习。文件围绕数据质量检查与预处理设计可帮助读者在接近真实的场景中依次演练缺失值处理、异常值检测、一致性校验、类型转换与去重合并等核心步骤。压缩包内共11个文件涵盖SQL脚本、CSV、JSON、XML、XLS/XLSX及TXT等格式SQL脚本适合导入数据库进行多表查询与清洗CSV和Excel便于使用Pandas、Excel或R开展表格处理JSON与XML用于半结构化数据解析TXT则贴近日志与文本清洗场景包体仅96KB轻量便携便于课堂分发与快速实验。目前已有528人学习下载适合高校大数据课程、企业培训及个人自学。通过实际操作这些数据源可以掌握从多源数据导入、脏数据识别到规范化输出的完整流程积累Pandas、SQL或ETL工具的真实使用经验为后续分析与建模打好数据基础。1. 拿到压缩数据源后的第一道坎先把zip安全打开再说我敢打赌绝大多数人看到数据清洗数据源.zip这个文件名时第一反应是双击解压、拖出CSV、pd.read_csv()一把梭。但真正拿到真实数据源的人都会懂这个zip里藏着的东西往往比你想的乱得多子目录套子目录、文件名是韩文或日文乱码、某些文件还是加密的、甚至解压到一半直接报invalid zip archive: could not find eocd。这些坑我没少踩所以第一个章节就想把从zip到DataFrame这最后一段距离给讲透。1.1 用Python标准库操作zip而不是直接右键解压直接右键解压并没有错但当你面对的是数据清洗任务时我更推荐用Python内置的zipfile模块把解压这个动作写进脚本里。理由很朴素数据源是zip这件事本身说明你的上游交付方可能没有统一的数据管理规范zip里既有CSV又有JSON、既有数据文件又有说明文档你需要的是程序化的读取方式而不是每次手动点开去看。import zipfile from pathlib import Path data_path Path(data_source.zip) extract_dir Path(extracted_data) with zipfile.ZipFile(data_path, r) as zf: # 先把压缩包里的文件清单打出来心里有数 for info in zf.infolist(): print(info.filename, info.file_size) zf.extractall(extract_dir)这一步最核心的价值是让你先看到zip内部的完整文件结构。我习惯把infolist()的结果打印出来因为真实场景里zip里往往混着__MACOSX这种系统残留目录、乱七八糟的临时文件甚至一份数据同时出现v1和v2两个版本。先看清再动手能省掉后面大量排查时间。1.2 文件名乱码和加密zip两个高频意外关于乱码提一次就够痛了非UTF-8编码的文件名在解压后显示为乱码这是zipfile模块的经典问题。zip格式本身对文件名编码没有强制约定Windows下很多压缩工具生成的是GBK编码而Python 3的zipfile默认按UTF-8解码所以解压出来一堆乱码文件名。我常用的处理思路是不直接解压到磁盘而是把文件名先重编码。具体做法是filename.encode(cp437).decode(gbk)。这里稍微解释一下原理zipfile解不出正确文件名时很多情况下文件名的原始字节被错误地按cp437解码成了Unicode字符串所以你需要先把它还原成原始字节再用正确的中文编码重新解码。这个技巧在批量处理国内业务线上游交付的数据包时尤其实用。至于加密zip首先得明确一点破解别人加密的压缩包不是本文讨论的事。但如果是自己的数据源包密码忘了或者交接时没拿到那就很现实了。Python处理加密zip的标准库能力非常有限zipfile对传统ZipCrypto加密格式只能读不能写更别提AES加密的包。我在这种情况下一般直接用pyzipper这个第三方库它对AES加密的zip支持很好import pyzipper with pyzipper.AESZipFile(data_source.zip) as zf: zf.setpassword(byour_password) zf.extractall(extracted_data)这里想多说一句如果密码真的彻底丢了市面上有些专业工具可以做密码恢复但那是暴力破解的思路成本极高。与其花时间破解不如联系上游重新获取密码这是我的真实建议。1.3 压缩包损坏时先分清是文件问题还是代码问题热搜词里有一条很典型导入失败caused by: invalid zip archive: could not find eocd。这个报错的意思是zip文件的末尾找不到End of Central Directory记录简单的说这个文件在传输或下载过程中被截断了或者根本不是一个完整的zip文件。遇到这个错别急着改代码先验证文件本身python -c import zipfile; zipfile.ZipFile(data_source.zip).testzip()如果输出None说明压缩包完整性没有问题问题多半出在读取逻辑上。如果确实报错那就没辙了重新下载或让上游重新生成是唯一的解。检查完文件完整性再检查你的读取代码有没有问题这个顺序千万别反过来否则你会拿着好端端的zip折腾半天代码最后发现是文件本身烂了。2. pandas清洗的四个主战场缺失、异常、重复、格式zip打开之后真正的重头戏才开始。数据清洗这个词听起来宽泛落到实际操作中无非就是跟缺失值、异常值、重复值、格式不一致这四个问题纠缠。很多人一上来就是dropna()和drop_duplicates()跑完看眼head就觉得干净了——这种清洗思路大概率要翻车。我整理了四类高频场景的完整处理思路每一个都是在真实项目中验证过的。2.1 缺失值处理不是dropna一个选项走天下dropna()绝对不是处理缺失值的第一选择甚至不是第二选择。真实数据里的缺失值分布是有规律的某个字段缺了20%、另一个字段缺了80%有的缺失是随机发生的有的缺失则跟业务强相关比如用户没填收入字段可能是因为他不想填而非没有。dropna()一刀切删行会把有效信息一起删掉。我建议的排查流程是missing_stats df.isna().mean().sort_values(ascendingFalse) print(missing_stats[missing_stats 0])先看清每个字段的缺失比例和分布再决定策略缺失比例低于5%的字段直接删除缺失行问题不大缺失比例在30%到50%之间的要考虑用中位数、众数或基于其他列的预测值来填充缺失比例超过70%的字段大概率要整列删掉因为它已经不具备分析价值了。填充方式的选择也有讲究。数值型字段用中位数比用均值更稳健因为均值容易被极端值带偏离散型字段用众数填充时要注意如果众数占比本身就低填充反而会引入噪声。还有一种场景比较特殊——时间序列数据里的缺失值用ffill或interpolate比用统计值填充合理得多因为相邻时刻的值具有天然的连续性。2.2 异常值不能只看describe要结合业务和分布df.describe()能看到最大值、最小值、均值、标准差但异常值从来不是靠观察这些统计量就能发现的。比如一个用户年龄字段最大值是150mean被拉到45看起来数据分布很均匀但150这个值放现实里就是不可能的。我处理异常值的套路分三步。第一步画出分布图确认是否有明显偏离的值df.boxplot()是最快的可视化方式。第二步针对每个字段问一个业务问题这个值的合理范围是什么用户年龄在0到120之间、交易金额大于0、评分在1到5分之间。第三步才是用技术手段处理超范围的值可以直接置空再用上一节的缺失值策略填充符合范围但明显偏大的值比如某用户一个月消费了99万这个要单独拎出来可能是大客户也可能是数据录入多加了一个0不能一刀切删掉。这里有个经验清洗异常值之前先在原始数据上做一个数据血缘标记把哪些行被标记为异常、为什么标记为异常记录下来。这么做的好处是后续分析时如果发现数据波动异常你能追溯回清洗逻辑而不是对着一个清洗后的死数据干瞪眼。2.3 重复值去重关键在于确定重复的定义drop_duplicates()默认找完全一样的行但真实业务里完全重复的数据其实很少见常见的是部分字段重复。比如一个用户在两行里user_id相同但地址字段一个填了、一个没填这两行并不能用drop_duplicates()直接处理需要你手动决定保留哪一行。我处理这类问题的思路是先指定去重的主键再用subset参数控制哪些列参与判断最后用keep参数决定保留哪一条。df_cleaned df.drop_duplicates(subset[user_id, order_time], keepfirst)还有一点容易被忽略去重之前先对数据排序。如果你用keepfirst保留的是排序后第一个出现的记录而排序顺序会直接决定哪条记录被保留。所以正确姿势是先按时间戳或数据质量排序再去重df df.sort_values(record_time, ascendingFalse) df df.drop_duplicates(subset[user_id], keepfirst)这样能保证保留的是每个用户最新的一条记录而不是随机的一条。2.4 文本字段的格式清洗最容易被低估数值和日期格式的问题大家都重视文本字段的清洗却经常被扔到后面结果后面做join或统计时疯狂报错。最典型的几个坑字段首尾有空格或不可见字符、全角半角数字混用、英文字母大小写不统一、手机号或身份证号被Excel自动转成了科学计数法。我通常在清洗流程里加这样一个函数def clean_text_series(s): s s.astype(str).str.strip() s s.str.replace(r\s, , regexTrue) s s.str.fullwidth_to_halfwidth() # 全角转半角 return s注意\s这一步很多从系统导出的数据字段里藏着肉眼看不见的换行符或制表符不清理干净后续拿这个字段做groupby同一个值会被切分成好几种看起来一样但实际不同的类别。日期字段的格式清洗也提一句多个数据源合并时常出现2024/01/01、2024-01-01、20240101三种格式混用的情况。保险的做法是统一用pd.to_datetime()转换遇到无法解析的值设置errorscoerce让它变成NaT再走缺失值处理流程。这一步做晚了后面所有依赖时间字段的聚合都会出错。3. 多个zip、多数据源的合并策略从混乱到一致的必经之路热搜词里有多数据源和springbootmybatisplus多数据源这些词看得出来很多人关心的是怎么处理来自多个地方的数据。放在数据清洗这个场景下就是你有两个zip甚至三个zip每个zip里是一份不同维度的数据这已经不是简单的读一个文件的问题了而是怎么把多份数据按规则拼成一个可用的整体。3.1 先统一路径读取再谈合并当多个zip文件里的表结构不一致时第一步不是急着concat而是先做一个数据仓库式的登记遍历所有zip读出每个文件的表头记录字段名和数据类型形成一份数据字典。def inspect_zip(zip_path): with zipfile.ZipFile(zip_path, r) as zf: csv_files [f for f in zf.namelist() if f.endswith(.csv)] for name in csv_files: with zf.open(name) as f: df pd.read_csv(f, nrows5) print(name, df.columns.tolist())这一步的作用是让你在做合并之前就知道各张表的字段差异而不是在merge报错时才开始排查。多数据源场景里最常见的坑是两张表里看似同一个字段的命名不一样一张表叫user_id另一张表叫userId还有一张表叫用户ID。这种不一致如果不提前梳理清楚等到项目后期再发现清洗逻辑就得推倒重来。3.2 merge的键设计数据质量决定了合并成败多源合并的核心是join key。先把各个数据源中用户标识字段的格式统一再检查有没有重复项——因为merge时如果键有重复会产生笛卡尔积行数会爆炸性增长。检查方法很简单df[df[user_id].duplicated()]如果这个结果不为空你要么先对键去重要么想清楚重复记录的业务含义。另一个值得重视的坑是键字段的脏数据user_id里混了空格、混了小数点10001和10001.0、字符型数字和数值型数字不一致。最常见的坑来源于Excel导出时长ID被转成了科学计数法原本的身份证号或订单号变成了1.23457E17精度直接丢了。遇到这种状况任何merge都白搭因为位数都对不上。我查过太多的线上事故起因就是两个系统导出的ID字段一个变成了浮点、一个是字符串。处理方案是把ID字段统一转换成字符串并去掉末尾的.0def clean_id(s): if s.dtype float64: s s.astype(Int64).astype(str) elif s.dtype object: s s.astype(str).str.replace(r\.0$, , regexTrue) s s.str.strip() return s键干净了merge才有意义这是多源数据清洗里最基本的常识。3.3 合并后的一致性校验不检查等于白干合并完成不代表结束还得做一轮一致性校验。我当时做两个数据源合并时合并前后行数变化、键的唯一性、关键字段的缺失率变化这三项都会跑一遍对比脚本。比如合并前A表有10万行、B表有8万行合并后如果是15万行说明有5万行没匹配上那就要回去查这5万行的键到底为什么没对上。合并后如果超过80万行基本可以确认键有重复导致笛卡尔积赶紧回去修键。这种“结果合理性判断”的能力是区分初级数据工作者和资深数据工作者的重要分水岭。数据清洗不是说把代码跑通了就完事而是要能解释清为什么这个数字是这个数字。4. 把清洗脚本沉淀成可复用的数据管道数据清洗这件事只要跑一次往往就够让人怀疑人生了。但如果你负责的是周期性更新的数据源比如每周从业务方拿到一个新的zip那你就不能每次重新手写一遍清洗脚本——你要做的是把清洗逻辑沉淀成一个可复用的管道争取让下一次清洗变成丢进去一个zip吐出来一份干净数据的事。4.1 清洗函数要按阶段拆分别写成一个巨型函数我最开始写清洗脚本时习惯把所有步骤堆在一个函数里读文件、删缺失、去重、改格式、合并、导出一气呵成。这个写法的问题是当某个环节出错时你很难定位是哪里出了问题而且如果数据源的格式稍有变化整个函数都得改。后来我改成按阶段拆分每个阶段只做一件事def load_zip_to_df(zip_path): pass def remove_invalid_rows(df): pass def fill_missing_values(df): pass def normalize_text_fields(df): pass def deduplicate_records(df): pass def save_clean_data(df, output_path): pass每个函数之间有清晰的输入输出协议输入DataFrame输出DataFrame不修改外部状态。这样你在调试时可以单独调用任意一个函数直接观察它对数据的影响。跑完整条管道后如果发现最终结果不对劲可以往上一层层倒推看是哪一步产生的偏差。4.2 中间结果落盘不是浪费时间而是买保险处理大规模或来源复杂的清洗任务时我不会直接一条流水线跑到底而是会在关键步骤之后把中间结果落盘保存。比如缺失值填充后保存一份、去重后保存一份。这么做有两个好处一是当最终结果有问题时能准确知道是哪一步引入的问题二是当数据源更新后你只需要重跑变化之后的部分而不需要从头到尾再执行一遍。当然中间文件会占用磁盘空间所以我会在脚本最后加一步自动清理逻辑只保留必要的时间戳快照和最新的干净数据集。4.3 数据源是周期性更新时要设计增量而不是全量重跑如果数据源每周都会更新你面对的就不是一个静止的zip而是一个持续演进的zip。全量重跑在这种场景下效率很低而且会把上一轮清洗后人工处理过的数据覆盖掉。比较稳妥的方案是引入时间戳做增量处理每次处理完数据后记录当前数据源zip的修改时间或文件哈希值。下次运行脚本时先对比这个标记如果zip没有变化就跳过清洗如果变了就只处理新增的部分。这个方案不复杂但对长期维护数据管道来说能省下大量的重复劳动。import hashlib def hash_file(path): h hashlib.md5() with open(path, rb) as f: for chunk in iter(lambda: f.read(4096), b): h.update(chunk) return h.hexdigest()先把文件的哈希值算出来和上次记录的做对比再决定要不要全量重跑。逻辑简单但很多人想不到用这个办法。4.4 一个小型的可复用管道示例最后给一个我在小项目中常用的管道骨架整体思路就是函数拆分阶段落盘关键指标留日志def run_clean_pipeline(zip_path: str, output_path: str) - None: df load_zip_to_df(zip_path) df remove_invalid_rows(df) df fill_missing_values(df) df normalize_text_fields(df) df deduplicate_records(df) save_clean_data(df, output_path) print(foriginal rows: {len(df)}, cleaned rows: {len(df)}) print(fremaining missing rate:\n{df.isna().mean()[df.isna().mean() 0]})这个骨架看起来简单但真正用起来有几个细节值得注意remove_invalid_rows和fill_missing_values这两个函数的顺序直接决定最终数据的质量——如果你先填充缺失值再删无效行等于在无效数据上做了无意义的填充反过来先删无效行再填充能避免脏数据影响填充统计量。清洗步骤的执行顺序本身就是一种经验这些内容文档里不会明说只有踩过坑你才会真正重视起来。我自己在跑这种管道时习惯在跑完那行打印里多加一个维度清洗前后行数变化原因明细。删了多少行是缺失导致的、多少行是异常导致的、多少行是重复导致的都分别统计出来。等到向业务方交数据时这些统计能直接回答数据为什么少了这么多可以省掉很多沟通成本。最后再分享一个小技巧清洗完的数据在导出给下游之前先抽样10行左右人工看一眼确认没有出现整列变成NaN或者文本字段错乱的情况。我在实际项目中碰到过一次清洗逻辑把某个中文字段里的顿号误判成了分隔符整列被切开要不是抽样检查及时发现这一份错误数据就会一路畅通地进入下游报表造成的返工成本远高于清洗本身。本文还有配套的精品资源点击获取