上个星期我又踩了一次数据清洗的老坑。手里一份三十多万行的渠道商名单CSV从两个系统导出后直接拼接然后我习惯性地先做了个整行去重。等统计完给业务方一看对面直接说数不对——同一个门店在A系统里联系电话是“138-0000-0000”在B系统里是“13800000000”中间的分隔符不一样我按整行去重根本识别不出这是同一条记录。那天下午我基本就耗在返工上了。后来我把这套处理流程重新理了一遍总结出一个很朴素的结论CSV清洗的正确顺序应该是先做标准化再做去重最后把每一步操作都留下可检查的报告。标准化不到位去重就只是走过场去重不留报告后面谁也没法确认你到底删了什么、凭什么删。这篇文章我把完整的思路、代码和实际踩过的坑都写出来给同样需要处理本地CSV、又不想在细节上翻车的人做个参考。无论你是用pandas、SQL还是纯命令行工具这个处理顺序和检查逻辑都是通用的。1. 先把顺序理清楚标准化前置去重才有意义很多人在处理CSV时的第一反应就是“我要去重”这个直觉没有错但如果数据本身还没有统一表达形式去重这个动作就无从谈起。去重的本质是判断两条记录是否在某个特征维度上等价而“等价”的前提是特征的表达方式一致。先标准化、再去重本质上是让比较的基准先成立。1.1 同一个商户为什么在名单里出现了两次我给你构造一个极常见的例子。下面这张表只有三个字段门店名称、联系电话、合作状态。门店名称联系电话合作状态天河店138-0000-0000正常天河店13800000000正常天河店138 0000 0000正常肉眼看一眼就知道这是同一个店三条记录说的是同一件事。但如果直接跑整行去重这三行会因为字符串级别的差异被判定为互不相同全部保留下来。原因很简单138-0000-0000、13800000000、138 0000 0000在字符层面确实不是同一个值。类似的场景还有日期。2024/01/05、2024-01-05、20240105、2024年1月5日在语义上是同一天但在CSV文件里是四种完全不同的字符串。如果不先把日期统一成一种格式去重时就等于睁着眼睛漏掉重复数据。所以标准化的核心作用是什么是把“语义相同但形态不同”的值收敛成同一种形态让后续去重能在统一的基准上进行。这一步不做后面所有的去重逻辑都是在沙地上建楼。1.2 标准化不到位的“假唯一”比重复更坑重复数据容易发现因为至少你看得出记录多了一行。“假唯一”才是更隐蔽的问题——它披着“每条都不一样”的外衣实际上在业务层面是同一条实体。这种问题不会让文件看起来异常但会让后续的统计结果整体虚高。举一个我实际遇到过的例子。我曾经清洗过一份会员数据按“邮箱”字段去重后保留了近十万条记录。后来业务方做营销前抽样检查发现里面有不少邮箱看起来“差不多”比如ZhangSangmail.com和zhangsangmail.com、zhangsangmail.com和zhangsangmail.com注意末尾空格。这些记录在去重时全部被保留因为字符串不完全相等。可它们其实是同一个人的同一个邮箱。这就是“假唯一”的破坏力统计数据看起来没问题业务执行时才发现名单里有大量重复的触达对象。更麻烦的是如果标准化滞后文件已经被下游系统导入再回头修数据就要牵扯一堆联动。所以要控制清洗质量不是看去重后发现多少重复而是看标准化后还能不能发现新的重复。我个人的习惯是标准化之后专门跑一次“重复率对比”看看同一批数据在标准化前和标准化后重复行数分别是多少。如果标准化后的重复数量明显增加说明前面的标准化工作起到了真正的收敛作用如果几乎没变化那就要怀疑是不是某些字段没有处理干净。1.3 用SQL做清洗时顺序同理有人可能觉得这套逻辑只适合pandas这类数据处理工具其实用SQL做CSV清洗时顺序也一样。SQL里最常见的去重写法是SELECT DISTINCT store_name, phone FROM raw_table;或者用窗口函数SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY store_name, phone ORDER BY updated_at DESC) AS rn FROM raw_table ) t WHERE rn 1;问题在于如果表里的phone字段有的带横线、有的不带有的首尾带空格上述SQL在去重时依然会把同一个人当成两个不同的人。所以在SQL里正确的做法是先做一步标准化比如SELECT TRIM(store_name) AS store_name, REGEXP_REPLACE(phone, [^0-9], ) AS phone FROM raw_table;等字段形态统一了再放进DISTINCT或ROW_NUMBER()里去重。逻辑跟pandas完全一样只是工具不同。2. 标准化的落地清单编码、字符串、空值、日期与数值都要管标准化到底要做哪些事这是我被问得最多的一个问题。很多人的理解是“把数据变好看一点”实际上标准化的对象非常具体可以拆成编码与表头、字符串清理、空值统一、日期与数值归一这几大块。每一块都有对应的坑。2.1 编码和表头先在读取阶段解决一半问题CSV文件最常见的一个坑是编码问题。用Excel导出的CSV很多是GBK或GB18030编码用pandas直接按默认UTF-8读取出来的就是乱码。还有一种情况是读取时是好的但写完CSV后再用Excel打开变成乱码这通常是因为缺少BOM头。处理这个问题最省心的办法是在读文件时显式指定编码并且在写文件时带上utf-8-sigimport pandas as pd # 读取时指定编码避免乱码 df pd.read_csv(raw_data.csv, dtypestr, encodingutf-8) # 如果上面的编码不对试试 gbk # df pd.read_csv(raw_data.csv, dtypestr, encodinggbk) # 写出时用 utf-8-sig这样 Excel 打开不会乱码 df.to_csv(clean_data.csv, indexFalse, encodingutf-8-sig)这里有个小经验dtypestr非常重要。CSV里的电话号码、订单号这类字段如果不强制指定为字符串pandas会把它们转成数字导致前导零丢失、长数字变成科学计数法。比如订单号0123456789会被读成123456789这种数据一旦进入去重环节就会产生“看起来相同其实被截断”的假重复或者反过来丢掉有效信息。所以凡是长度超过15位、或者可能带前导零的字段我建议一律用dtypestr读取。表头标准化也是容易被忽略的环节。同一份数据在不同系统里导出的列名可能不一样比如StoreName和store_name、店铺名称和门店名称如果不去统一后续代码写起来会很痛苦。我一般会在读取后先做一次列名归一df.columns [col.strip().lower().replace( , _) for col in df.columns]把列名统一成小写下划线风格后后面所有代码都不会因为列名大小写不一致而踩坑。2.2 字符串和空值把“看不见的差异”处理掉字符串清洗是标准化中最细碎的部分也是最容易漏的部分。下面这些场景都是我在真实数据里碰到过的字符串首尾有空格肉眼看不出来但strip()之前它就是和没有空格的字符串不相等全角空格\u3000常见于中文输入法环境下编辑过的CSVstr.strip()默认处理不了括号有的是英文括号()有的是中文括号数字和单位之间有时有空格有时没有。处理这些“看不见的差异”我建议写一个统一的清理函数把每一列都过一遍def clean_string_series(s): return (s .astype(str) .str.strip() # 去掉首尾半角空格 .str.replace(\u3000, , regexFalse) # 去掉全角空格 .str.replace(, (, regexFalse) .str.replace(, ), regexFalse) .str.replace(r\s, , regexTrue) # 去掉中间多余空白按需使用 )注意str.replace(r\s, )这步要谨慎。电话号码这类字段字中间的空白确实应该去掉但地址字段如果去掉所有空白可读性会变差。更合理的做法是对不同列使用不同策略身份证号、电话号、订单号这种标识类字段做“去所有空白”处理名称、地址这类字段只做首尾去空格和全角转半角。空值也是个老大难。在CSV里“空”的表达方式太多了单元格完全空着、字符串NULL、NA、N/A、None、-、未知甚至四个空格。如果不统一后面做统计或者去重时每个值都会被当成不同的东西。我的做法是先统一成一个标记再按业务需求决定是填空还是丢弃na_placeholders [, NULL, NA, N/A, None, null, -, nan] df df.replace(na_placeholders, pd.NA) # 如果业务上需要空值参与比较可以先填充一个统一占位符 df df.fillna()这里有一个关于去重的坑值得单独说pandas里的drop_duplicates()会把所有NaN值视为同一个值。如果某一列大量缺失做整行去重时这些缺失行内部会被互相判定为重复。这个问题我会在第3章详细展开但标准化的意义就在于先把缺失值处理成一个明确的标记再去重你才能准确知道“重复”到底是发生在有效数据上还是被缺失值搞出来的假象。2.3 日期与数值统一避免同值不同形日期字段是“同值不同形”的重灾区。一份CSV里同一个时间点可能被写成2024/01/05 10:30:00也可能被写成2024-01-05还可能是Excel导出的序列号。要统一最省心的是交给pandas.to_datetime解析后格式化成同一种字符串df[signup_date] pd.to_datetime(df[signup_date], errorscoerce) df[signup_date] df[signup_date].dt.strftime(%Y-%m-%d)这样无论源数据里是哪种格式最后都会统一成YYYY-MM-DD。要注意的是errorscoerce会把无法解析的日期变成NaT这些坏数据会在后续检查时被单独找出来而不是默默参与去重。数值字段也类似。有些CSV里的金额是1,234.56带千分位有些是1234.56有些甚至带货币符号¥1,234.56。去重前如果不把这些统一成纯数字1,234.56和1234.56就会被当成两个不同值。处理方法是用正则把非数字字符去掉def clean_numeric_series(s): return (s .astype(str) .str.replace(r[^0-9.\-], , regexTrue) .replace(, pd.NA) .astype(float) )这块我的建议是数字字段的可读性让位给可比性。清洗阶段的目标是让相同语义的值在字符层面完全一致美观是后面展示环节的事。2.4 标准化代码的推荐写法把上面这些逻辑串成一个函数是提升复用率的关键。我习惯一次性把常见的标准化步骤封装起来每接一份新CSV先跑一遍通用清理再做针对性的字段处理def normalize_dataframe(df): df df.copy() # 1. 列名标准化 df.columns [str(c).strip().lower().replace( , _) for c in df.columns] # 2. 字符串列统一清理 str_cols df.select_dtypes(include[object]).columns for col in str_cols: if col not in [signup_date, amount]: # 排除需要单独处理的列 df[col] (df[col] .astype(str) .str.strip() .str.replace(\u3000, , regexFalse) .str.replace(, (, regexFalse) .str.replace(, ), regexFalse)) # 3. 空值占位符统一 df df.replace([, NULL, NA, N/A, None, null, -, nan], pd.NA) return df这套方案最大的好处是同一份CSV不管跑多少次清洗规则都是一样的可复现、可审计。不会出现这次清洗和上次清洗结果不一致的情况。3. 去重的关键决策按什么子集、保留哪一行、怎么防范NaN坑标准化做完之后去重才有意义。但去重也不是简单调一个drop_duplicates()就完事里面还牵扯到三个关键决策按哪些列判断重复、保留哪一行、缺失值怎么处理。3.1 整行去重还是按字段子集去重drop_duplicates()默认是整行去重也就是只有所有列的值完全一样才判定为重复。这个逻辑听上去没问题但实际业务里很少用整行去重因为不同系统导出的同一份数据很可能存在“某个辅助字段不一致、但主键一致”的情况。比如同一个订单A系统里备注字段是空的B系统里备注写了“已电话确认”这时候两行在整行级别不重复但在业务逻辑里它们就是同一个订单。所以更常见的做法是按业务主键或关键字段去重df df.drop_duplicates(subset[store_name, phone_clean], keepfirst)subset参数选哪些字段取决于你对业务的理解。这个选择宁可偏严再放开也不要一开始就把唯一性建立在“所有字段都一致”上。通常我会先按最核心的标识字段去重比如身份证号、手机号、订单号结果里如果发现有价值的记录被误删再通过报告排查出来调整subset。3.2 先排序再去重保留哪一行不能靠运气drop_duplicates(keepfirst)默认保留首次出现的行keeplast保留最后出现的行。问题在于如果你的数据没有经过排序所谓“首次出现”就是源文件里的顺序而这个顺序在大多数情况下没有任何业务含义。举个实际场景。会员表里有两条记录会员ID相同但一条是刚注册时的信息一条是最近更新过的信息。如果直接drop_duplicates(subset[member_id], keeplast)谁在文件里靠后谁就被保留如果源文件顺序是随机的那保留哪条就完全不可控了。正确的姿势是先用一个有业务意义的字段排序再去重df (df .sort_values(updated_at, ascendingFalse) # 把最新更新的记录排前面 .drop_duplicates(subset[member_id], keepfirst) # 保留最新的 .sort_index() # 恢复顺序看心情决定要不要 )这样去重结果就变得可解释了我们保留的是每个会员ID中更新时间最新的一条。这个逻辑写进报告里任何人都能看懂、能复现。SQL里对应的做法是前面写过的SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY member_id ORDER BY updated_at DESC) AS rn FROM member_table ) t WHERE rn 1;逻辑完全一致先排序再分组去重。3.3 缺失值在去重中的隐性行为这个坑我专门拎出来讲因为太多人在实践里栽过。pandas的drop_duplicates()在处理NaN时有一个特殊行为所有NaN被看作是同一个值。这意味着如果某列有大量缺失而这些行又恰好被纳入了去重子集那它们会被批量判定为重复。举个具体例子。按phone字段去重表中有一千行是phoneNaN那这一千行会先内部互相判重最终只保留一行剩下的九百九十九行全被删掉。如果这些缺失电话的行在业务上是不同的客户那这就不是去重而是误删了。解决思路有三种在标准化阶段就把空值统一填充成一个明确的占位符例如UNKNOWN这样不会影响去重判断将缺失值从去重判断中排除先处理有值数据的去重再单独处理缺值数据使用subset时选择缺失率较低的字段避免大量缺失列参与判重。我通常选第一种因为标准化阶段反正要做缺失值统一干脆就统一成有业务含义的占位符而不是让NaN这种隐式语义在后续操作里捣乱。3.4 用行级哈希指纹做快速判重除了字段级判重还有一种场景需要做“整行镜像”检测文件合并时同一行被完整地复制了两遍。这种行只要所有列的都相等就是重复但逐列比较在列很多时挺麻烦。这时候可以入行级哈希指纹。简单实现是用hashlib对整行拼接结果做MD5给每一行打一个指纹再基于指纹去重import hashlib def row_fingerprint(row): raw |.join([str(v) for v in row]) return hashlib.md5(raw.encode(utf-8)).hexdigest() df[_fingerprint] df.apply(row_fingerprint, axis1) df df.drop_duplicates(subset[_fingerprint], keepfirst)pandas也提供了更方便的工具from pandas.util import hash_pandas_object df[_hash] hash_pandas_object(df, indexFalse).astype(str)指纹列的好处是它把整行信息压缩成一个短字符串去重判断非常快而且指纹本身可以写进报告作为检查依据。要注意的是指纹计算前必须先完成字段标准化否则同样的数据因为格式不同会生成不同指纹等于白做。4. 给清洗留证据删除明细、汇总报告与文件指纹很多人做完去重后直接to_csv输出干净文件就算完事。但如果后续业务方问一句“你到底删了哪些删之前是多少行”你拿不出证据那就尴尬了。更现实的情况是某条重要的记录被误删了你需要在几分钟内定位到它是为什么被删的。所以可检查的报告不是锦上添花是清洗流程的必要环节。4.1 删除明细报告每一行被删的数据都要留底删除明细的基本思想很简单把判定为重复、且未被保留的那些行单独导出一个CSV文件。文件里除了原始数据建议再加上几个说明性字段方便回溯字段说明_dup_group重复组编号同一组重复记录共用一个编号_is_kept该行是否被保留_drop_reason删除原因例如“按member_id判定重复保留最新记录”_original_index原文件中的行号/索引方便回到源文件定位生成删除明细的做法是这样# 在标准化、排序后的全量数据上打上重复标记 df_all[_dup_flag] df_all.duplicated(subset[member_id], keepFalse) # 去重保留下来的数据 df_kept df_all.drop_duplicates(subset[member_id], keepfirst) # 被删数据 参与重复的行中不在保留集合里的行 removed_index df_all.index.difference(df_kept.index) df_removed df_all.loc[removed_index].copy() df_removed[_drop_reason] member_id重复保留更新时间最新记录这里用duplicated(subset..., keepFalse)先标出所有参与重复的行再用保留集合和被删除集合做索引差逻辑很清晰。最后把删除明细写出去df_removed.to_csv(output/removed_records.csv, indexFalse, encodingutf-8-sig)有了这个文件任何一行数据被删都能查得到它原来在第几行因为跟谁重复被谁替代了。这比光给一个干净文件要可靠得多。4.2 清洗前后指标对比用数字证明清洗有效除了删除明细我还会生成一份“清洗前后指标对比”把核心数字汇总出来。常用的指标包括原始行数、标准化后行数、去重后行数被删除的行数关键字段的唯一值数量比如会员ID去重前多少、去重后多少每个字段的缺失值数量重复率即被删行数占原始行数的比例。这些指标可以整理成一个简单的文本或JSON报告summary { source_file: raw_channels.csv, total_rows_before: len(df_raw), total_rows_after_standardize: len(df_std), total_rows_after_dedup: len(df_kept), removed_rows: len(df_removed), duplicate_rate: round(len(df_removed) / len(df_raw), 4), unique_member_id_before: df_raw[member_id].nunique(), unique_member_id_after: df_kept[member_id].nunique(), } import json with open(output/summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2)这份报告的价值在于你不需要展开细节只看汇总数字就能判断这次清洗是否正常。比如重复率突然从5%飙到40%多半是标准化规则太激进或者subset选错了字段这时就能回到明细里去排查。4.3 文件MD5/SHA256指纹让报告和文件能对上“可检查”的最高级别是给输出的每个文件都生成一个唯一的指纹防止后续文件在传输、下载或转存过程中被无意改动。这就是很多人提到的“CSV文件怎么做MD5校验”。文件级别校验很简单直接对文件计算MD5或SHA256拿Linux/Mac终端来说md5sum output/clean_data.csv sha256sum output/removed_records.csvWindows PowerShell也可以用Get-FileHash -Algorithm SHA256 .\output\clean_data.csv把得到的哈希指纹写进报告文件以后任何人拿到这批CSV都可以重新计算哈希值和报告里的指纹比对确认文件是否被改动过。这有点类似给文件做了个“身份证号”把整个清洗工作的交付状态固化下来。对于行级别的MD5校验之前在第三章也提过就是用哈希指纹给每一行做唯一标记。两种指纹的应用场景不同文件级指纹用于验证交付文件的完整性行级指纹用于行级别的判重与追踪可以搭配使用。5. 一次完整的本地CSV清洗带注释的实战流程前面讲了理论和要点这一章我放一个可以直接照着改的完整示例。假设现在有两份渠道商CSV需要合并、清洗、去重并输出干净文件、删除明细和汇总报告。5.1 场景说明与目录结构先看目录结构project/ ├── raw/ │ ├── channel_a.csv │ └── channel_b.csv ├── output/ │ ├── clean_data.csv │ ├── removed_records.csv │ └── summary.json └── clean_csv.pychannel_a.csv和channel_b.csv字段基本一致但字段格式不完全相同可能有编码差异和重复记录。目标是把两份文件合并成一份干净数据按store_id去重保留最新更新的一条。5.2 分步代码与执行逻辑完整代码如下import pandas as pd import hashlib import json # ---------- 1. 读取与合并 ---------- df_a pd.read_csv(raw/channel_a.csv, dtypestr, encodingutf-8) df_b pd.read_csv(raw/channel_b.csv, dtypestr, encodingutf-8) # 如果列名不一致先统一字段名 df_a.columns [str(c).strip().lower().replace( , _) for c in df_a.columns] df_b.columns [str(c).strip().lower().replace( , _) for c in df_b.columns] df pd.concat([df_a, df_b], ignore_indexTrue) print(f合并后总行数: {len(df)}) # ---------- 2. 标准化 ---------- # 字符串列统一去掉首尾空格和全角空格统一中英文括号 str_cols df.select_dtypes(include[object]).columns for col in str_cols: if col not in [update_time, store_id]: df[col] (df[col] .astype(str) .str.strip() .str.replace(\u3000, , regexFalse) .str.replace(, (, regexFalse) .str.replace(, ), regexFalse)) # 门店ID如果存在非数字字符一律去除 if store_id in df.columns: df[store_id] df[store_id].astype(str).str.replace(r\D, , regexTrue) # 日期字段统一为 YYYY-MM-DD if update_time in df.columns: df[update_time] pd.to_datetime(df[update_time], errorscoerce).dt.strftime(%Y-%m-%d) # 空值占位符统一 df df.replace([, NULL, NA, N/A, None, null, -, nan], pd.NA) df df.fillna() # ---------- 3. 生成行级指纹可选用于快速核对 ---------- def row_fingerprint(row): raw |.join([str(v) for v in row]) return hashlib.md5(raw.encode(utf-8)).hexdigest() df[_row_fingerprint] df.apply(row_fingerprint, axis1) # ---------- 4. 去重先按更新时间排序再按门店ID去重 ---------- df_sorted df.sort_values(update_time, ascendingFalse, na_positionlast) df_kept df_sorted.drop_duplicates(subset[store_id], keepfirst) # 标记所有参与重复的行 df[_dup_flag] df.duplicated(subset[store_id], keepFalse) removed_index df.index.difference(df_kept.index) df_removed df.loc[removed_index].copy() df_removed[_drop_reason] store_id重复保留update_time最新记录 print(f去重后行数: {len(df_kept)}删除行数: {len(df_removed)}) # ---------- 5. 输出文件 ---------- df_kept.drop(columns[_dup_flag]).to_csv(output/clean_data.csv, indexFalse, encodingutf-8-sig) df_removed.drop(columns[_dup_flag]).to_csv(output/removed_records.csv, indexFalse, encodingutf-8-sig) # ---------- 6. 汇总报告 ---------- summary { source_files: [channel_a.csv, channel_b.csv], total_rows_after_merge: len(df), total_rows_after_dedup: len(df_kept), removed_rows: len(df_removed), duplicate_rate: round(len(df_removed) / len(df), 4), unique_store_id_before_dedup: df[store_id].nunique(), unique_store_id_after_dedup: df_kept[store_id].nunique(), } with open(output/summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2) print(清洗完成结果保存在 output/ 目录下)这段代码的逻辑很直白读文件、合并、标准化、打指纹、排序去重、输出明细和报告。你可以根据实际字段名调整subset和排序字段框架不需要动。5.3 跑完之后的检查清单代码能跑通只是第一步我每次清洗完还会做下面这套检查确保结果真的可信看summary.json里的删除行数和重复率如果异常偏高先怀疑标准化阶段是不是把大量有效值误统一成了同一个占位符打开removed_records.csv随机抽几行对照_drop_reason确认删除逻辑正确用文件哈希指纹再核对一遍输出文件的完整性确认没有写入时出现异常如果业务方需要可以把clean_data.csv和removed_records.csv一起交给他们而不是只给一个干净文件。这套检查动作做完我才会放心地把数据交付出去。最后分享一个我自己的习惯清洗脚本本身也要做版本管理每次跑批都在脚本头部写清楚清洗规则和日期输出报告的文件名加上时间戳比如summary_20250411.json。这样一个月后再有人问“这份数据是怎么洗出来的”你翻出当天的脚本、报告和指纹几分钟就能完整还原整个过程。清洗这件事最怕的不是数据脏而是脏完之后没人说得清到底做了什么。把过程留给别人查也留给未来的自己查这才是“可检查的报告”真正的意义。
上个星期我又踩了一次数据清洗的老坑。手里一份三十多万行的渠道商名单CSV从两个系统导出后直接拼接然后我习惯性地先做了个整行去重。等统计完给业务方一看对面直接说数不对——同一个门店在A系统里联系电话是“138-0000-0000”在B系统里是“13800000000”中间的分隔符不一样我按整行去重根本识别不出这是同一条记录。那天下午我基本就耗在返工上了。后来我把这套处理流程重新理了一遍总结出一个很朴素的结论CSV清洗的正确顺序应该是先做标准化再做去重最后把每一步操作都留下可检查的报告。标准化不到位去重就只是走过场去重不留报告后面谁也没法确认你到底删了什么、凭什么删。这篇文章我把完整的思路、代码和实际踩过的坑都写出来给同样需要处理本地CSV、又不想在细节上翻车的人做个参考。无论你是用pandas、SQL还是纯命令行工具这个处理顺序和检查逻辑都是通用的。1. 先把顺序理清楚标准化前置去重才有意义很多人在处理CSV时的第一反应就是“我要去重”这个直觉没有错但如果数据本身还没有统一表达形式去重这个动作就无从谈起。去重的本质是判断两条记录是否在某个特征维度上等价而“等价”的前提是特征的表达方式一致。先标准化、再去重本质上是让比较的基准先成立。1.1 同一个商户为什么在名单里出现了两次我给你构造一个极常见的例子。下面这张表只有三个字段门店名称、联系电话、合作状态。门店名称联系电话合作状态天河店138-0000-0000正常天河店13800000000正常天河店138 0000 0000正常肉眼看一眼就知道这是同一个店三条记录说的是同一件事。但如果直接跑整行去重这三行会因为字符串级别的差异被判定为互不相同全部保留下来。原因很简单138-0000-0000、13800000000、138 0000 0000在字符层面确实不是同一个值。类似的场景还有日期。2024/01/05、2024-01-05、20240105、2024年1月5日在语义上是同一天但在CSV文件里是四种完全不同的字符串。如果不先把日期统一成一种格式去重时就等于睁着眼睛漏掉重复数据。所以标准化的核心作用是什么是把“语义相同但形态不同”的值收敛成同一种形态让后续去重能在统一的基准上进行。这一步不做后面所有的去重逻辑都是在沙地上建楼。1.2 标准化不到位的“假唯一”比重复更坑重复数据容易发现因为至少你看得出记录多了一行。“假唯一”才是更隐蔽的问题——它披着“每条都不一样”的外衣实际上在业务层面是同一条实体。这种问题不会让文件看起来异常但会让后续的统计结果整体虚高。举一个我实际遇到过的例子。我曾经清洗过一份会员数据按“邮箱”字段去重后保留了近十万条记录。后来业务方做营销前抽样检查发现里面有不少邮箱看起来“差不多”比如ZhangSangmail.com和zhangsangmail.com、zhangsangmail.com和zhangsangmail.com注意末尾空格。这些记录在去重时全部被保留因为字符串不完全相等。可它们其实是同一个人的同一个邮箱。这就是“假唯一”的破坏力统计数据看起来没问题业务执行时才发现名单里有大量重复的触达对象。更麻烦的是如果标准化滞后文件已经被下游系统导入再回头修数据就要牵扯一堆联动。所以要控制清洗质量不是看去重后发现多少重复而是看标准化后还能不能发现新的重复。我个人的习惯是标准化之后专门跑一次“重复率对比”看看同一批数据在标准化前和标准化后重复行数分别是多少。如果标准化后的重复数量明显增加说明前面的标准化工作起到了真正的收敛作用如果几乎没变化那就要怀疑是不是某些字段没有处理干净。1.3 用SQL做清洗时顺序同理有人可能觉得这套逻辑只适合pandas这类数据处理工具其实用SQL做CSV清洗时顺序也一样。SQL里最常见的去重写法是SELECT DISTINCT store_name, phone FROM raw_table;或者用窗口函数SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY store_name, phone ORDER BY updated_at DESC) AS rn FROM raw_table ) t WHERE rn 1;问题在于如果表里的phone字段有的带横线、有的不带有的首尾带空格上述SQL在去重时依然会把同一个人当成两个不同的人。所以在SQL里正确的做法是先做一步标准化比如SELECT TRIM(store_name) AS store_name, REGEXP_REPLACE(phone, [^0-9], ) AS phone FROM raw_table;等字段形态统一了再放进DISTINCT或ROW_NUMBER()里去重。逻辑跟pandas完全一样只是工具不同。2. 标准化的落地清单编码、字符串、空值、日期与数值都要管标准化到底要做哪些事这是我被问得最多的一个问题。很多人的理解是“把数据变好看一点”实际上标准化的对象非常具体可以拆成编码与表头、字符串清理、空值统一、日期与数值归一这几大块。每一块都有对应的坑。2.1 编码和表头先在读取阶段解决一半问题CSV文件最常见的一个坑是编码问题。用Excel导出的CSV很多是GBK或GB18030编码用pandas直接按默认UTF-8读取出来的就是乱码。还有一种情况是读取时是好的但写完CSV后再用Excel打开变成乱码这通常是因为缺少BOM头。处理这个问题最省心的办法是在读文件时显式指定编码并且在写文件时带上utf-8-sigimport pandas as pd # 读取时指定编码避免乱码 df pd.read_csv(raw_data.csv, dtypestr, encodingutf-8) # 如果上面的编码不对试试 gbk # df pd.read_csv(raw_data.csv, dtypestr, encodinggbk) # 写出时用 utf-8-sig这样 Excel 打开不会乱码 df.to_csv(clean_data.csv, indexFalse, encodingutf-8-sig)这里有个小经验dtypestr非常重要。CSV里的电话号码、订单号这类字段如果不强制指定为字符串pandas会把它们转成数字导致前导零丢失、长数字变成科学计数法。比如订单号0123456789会被读成123456789这种数据一旦进入去重环节就会产生“看起来相同其实被截断”的假重复或者反过来丢掉有效信息。所以凡是长度超过15位、或者可能带前导零的字段我建议一律用dtypestr读取。表头标准化也是容易被忽略的环节。同一份数据在不同系统里导出的列名可能不一样比如StoreName和store_name、店铺名称和门店名称如果不去统一后续代码写起来会很痛苦。我一般会在读取后先做一次列名归一df.columns [col.strip().lower().replace( , _) for col in df.columns]把列名统一成小写下划线风格后后面所有代码都不会因为列名大小写不一致而踩坑。2.2 字符串和空值把“看不见的差异”处理掉字符串清洗是标准化中最细碎的部分也是最容易漏的部分。下面这些场景都是我在真实数据里碰到过的字符串首尾有空格肉眼看不出来但strip()之前它就是和没有空格的字符串不相等全角空格\u3000常见于中文输入法环境下编辑过的CSVstr.strip()默认处理不了括号有的是英文括号()有的是中文括号数字和单位之间有时有空格有时没有。处理这些“看不见的差异”我建议写一个统一的清理函数把每一列都过一遍def clean_string_series(s): return (s .astype(str) .str.strip() # 去掉首尾半角空格 .str.replace(\u3000, , regexFalse) # 去掉全角空格 .str.replace(, (, regexFalse) .str.replace(, ), regexFalse) .str.replace(r\s, , regexTrue) # 去掉中间多余空白按需使用 )注意str.replace(r\s, )这步要谨慎。电话号码这类字段字中间的空白确实应该去掉但地址字段如果去掉所有空白可读性会变差。更合理的做法是对不同列使用不同策略身份证号、电话号、订单号这种标识类字段做“去所有空白”处理名称、地址这类字段只做首尾去空格和全角转半角。空值也是个老大难。在CSV里“空”的表达方式太多了单元格完全空着、字符串NULL、NA、N/A、None、-、未知甚至四个空格。如果不统一后面做统计或者去重时每个值都会被当成不同的东西。我的做法是先统一成一个标记再按业务需求决定是填空还是丢弃na_placeholders [, NULL, NA, N/A, None, null, -, nan] df df.replace(na_placeholders, pd.NA) # 如果业务上需要空值参与比较可以先填充一个统一占位符 df df.fillna()这里有一个关于去重的坑值得单独说pandas里的drop_duplicates()会把所有NaN值视为同一个值。如果某一列大量缺失做整行去重时这些缺失行内部会被互相判定为重复。这个问题我会在第3章详细展开但标准化的意义就在于先把缺失值处理成一个明确的标记再去重你才能准确知道“重复”到底是发生在有效数据上还是被缺失值搞出来的假象。2.3 日期与数值统一避免同值不同形日期字段是“同值不同形”的重灾区。一份CSV里同一个时间点可能被写成2024/01/05 10:30:00也可能被写成2024-01-05还可能是Excel导出的序列号。要统一最省心的是交给pandas.to_datetime解析后格式化成同一种字符串df[signup_date] pd.to_datetime(df[signup_date], errorscoerce) df[signup_date] df[signup_date].dt.strftime(%Y-%m-%d)这样无论源数据里是哪种格式最后都会统一成YYYY-MM-DD。要注意的是errorscoerce会把无法解析的日期变成NaT这些坏数据会在后续检查时被单独找出来而不是默默参与去重。数值字段也类似。有些CSV里的金额是1,234.56带千分位有些是1234.56有些甚至带货币符号¥1,234.56。去重前如果不把这些统一成纯数字1,234.56和1234.56就会被当成两个不同值。处理方法是用正则把非数字字符去掉def clean_numeric_series(s): return (s .astype(str) .str.replace(r[^0-9.\-], , regexTrue) .replace(, pd.NA) .astype(float) )这块我的建议是数字字段的可读性让位给可比性。清洗阶段的目标是让相同语义的值在字符层面完全一致美观是后面展示环节的事。2.4 标准化代码的推荐写法把上面这些逻辑串成一个函数是提升复用率的关键。我习惯一次性把常见的标准化步骤封装起来每接一份新CSV先跑一遍通用清理再做针对性的字段处理def normalize_dataframe(df): df df.copy() # 1. 列名标准化 df.columns [str(c).strip().lower().replace( , _) for c in df.columns] # 2. 字符串列统一清理 str_cols df.select_dtypes(include[object]).columns for col in str_cols: if col not in [signup_date, amount]: # 排除需要单独处理的列 df[col] (df[col] .astype(str) .str.strip() .str.replace(\u3000, , regexFalse) .str.replace(, (, regexFalse) .str.replace(, ), regexFalse)) # 3. 空值占位符统一 df df.replace([, NULL, NA, N/A, None, null, -, nan], pd.NA) return df这套方案最大的好处是同一份CSV不管跑多少次清洗规则都是一样的可复现、可审计。不会出现这次清洗和上次清洗结果不一致的情况。3. 去重的关键决策按什么子集、保留哪一行、怎么防范NaN坑标准化做完之后去重才有意义。但去重也不是简单调一个drop_duplicates()就完事里面还牵扯到三个关键决策按哪些列判断重复、保留哪一行、缺失值怎么处理。3.1 整行去重还是按字段子集去重drop_duplicates()默认是整行去重也就是只有所有列的值完全一样才判定为重复。这个逻辑听上去没问题但实际业务里很少用整行去重因为不同系统导出的同一份数据很可能存在“某个辅助字段不一致、但主键一致”的情况。比如同一个订单A系统里备注字段是空的B系统里备注写了“已电话确认”这时候两行在整行级别不重复但在业务逻辑里它们就是同一个订单。所以更常见的做法是按业务主键或关键字段去重df df.drop_duplicates(subset[store_name, phone_clean], keepfirst)subset参数选哪些字段取决于你对业务的理解。这个选择宁可偏严再放开也不要一开始就把唯一性建立在“所有字段都一致”上。通常我会先按最核心的标识字段去重比如身份证号、手机号、订单号结果里如果发现有价值的记录被误删再通过报告排查出来调整subset。3.2 先排序再去重保留哪一行不能靠运气drop_duplicates(keepfirst)默认保留首次出现的行keeplast保留最后出现的行。问题在于如果你的数据没有经过排序所谓“首次出现”就是源文件里的顺序而这个顺序在大多数情况下没有任何业务含义。举个实际场景。会员表里有两条记录会员ID相同但一条是刚注册时的信息一条是最近更新过的信息。如果直接drop_duplicates(subset[member_id], keeplast)谁在文件里靠后谁就被保留如果源文件顺序是随机的那保留哪条就完全不可控了。正确的姿势是先用一个有业务意义的字段排序再去重df (df .sort_values(updated_at, ascendingFalse) # 把最新更新的记录排前面 .drop_duplicates(subset[member_id], keepfirst) # 保留最新的 .sort_index() # 恢复顺序看心情决定要不要 )这样去重结果就变得可解释了我们保留的是每个会员ID中更新时间最新的一条。这个逻辑写进报告里任何人都能看懂、能复现。SQL里对应的做法是前面写过的SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY member_id ORDER BY updated_at DESC) AS rn FROM member_table ) t WHERE rn 1;逻辑完全一致先排序再分组去重。3.3 缺失值在去重中的隐性行为这个坑我专门拎出来讲因为太多人在实践里栽过。pandas的drop_duplicates()在处理NaN时有一个特殊行为所有NaN被看作是同一个值。这意味着如果某列有大量缺失而这些行又恰好被纳入了去重子集那它们会被批量判定为重复。举个具体例子。按phone字段去重表中有一千行是phoneNaN那这一千行会先内部互相判重最终只保留一行剩下的九百九十九行全被删掉。如果这些缺失电话的行在业务上是不同的客户那这就不是去重而是误删了。解决思路有三种在标准化阶段就把空值统一填充成一个明确的占位符例如UNKNOWN这样不会影响去重判断将缺失值从去重判断中排除先处理有值数据的去重再单独处理缺值数据使用subset时选择缺失率较低的字段避免大量缺失列参与判重。我通常选第一种因为标准化阶段反正要做缺失值统一干脆就统一成有业务含义的占位符而不是让NaN这种隐式语义在后续操作里捣乱。3.4 用行级哈希指纹做快速判重除了字段级判重还有一种场景需要做“整行镜像”检测文件合并时同一行被完整地复制了两遍。这种行只要所有列的都相等就是重复但逐列比较在列很多时挺麻烦。这时候可以入行级哈希指纹。简单实现是用hashlib对整行拼接结果做MD5给每一行打一个指纹再基于指纹去重import hashlib def row_fingerprint(row): raw |.join([str(v) for v in row]) return hashlib.md5(raw.encode(utf-8)).hexdigest() df[_fingerprint] df.apply(row_fingerprint, axis1) df df.drop_duplicates(subset[_fingerprint], keepfirst)pandas也提供了更方便的工具from pandas.util import hash_pandas_object df[_hash] hash_pandas_object(df, indexFalse).astype(str)指纹列的好处是它把整行信息压缩成一个短字符串去重判断非常快而且指纹本身可以写进报告作为检查依据。要注意的是指纹计算前必须先完成字段标准化否则同样的数据因为格式不同会生成不同指纹等于白做。4. 给清洗留证据删除明细、汇总报告与文件指纹很多人做完去重后直接to_csv输出干净文件就算完事。但如果后续业务方问一句“你到底删了哪些删之前是多少行”你拿不出证据那就尴尬了。更现实的情况是某条重要的记录被误删了你需要在几分钟内定位到它是为什么被删的。所以可检查的报告不是锦上添花是清洗流程的必要环节。4.1 删除明细报告每一行被删的数据都要留底删除明细的基本思想很简单把判定为重复、且未被保留的那些行单独导出一个CSV文件。文件里除了原始数据建议再加上几个说明性字段方便回溯字段说明_dup_group重复组编号同一组重复记录共用一个编号_is_kept该行是否被保留_drop_reason删除原因例如“按member_id判定重复保留最新记录”_original_index原文件中的行号/索引方便回到源文件定位生成删除明细的做法是这样# 在标准化、排序后的全量数据上打上重复标记 df_all[_dup_flag] df_all.duplicated(subset[member_id], keepFalse) # 去重保留下来的数据 df_kept df_all.drop_duplicates(subset[member_id], keepfirst) # 被删数据 参与重复的行中不在保留集合里的行 removed_index df_all.index.difference(df_kept.index) df_removed df_all.loc[removed_index].copy() df_removed[_drop_reason] member_id重复保留更新时间最新记录这里用duplicated(subset..., keepFalse)先标出所有参与重复的行再用保留集合和被删除集合做索引差逻辑很清晰。最后把删除明细写出去df_removed.to_csv(output/removed_records.csv, indexFalse, encodingutf-8-sig)有了这个文件任何一行数据被删都能查得到它原来在第几行因为跟谁重复被谁替代了。这比光给一个干净文件要可靠得多。4.2 清洗前后指标对比用数字证明清洗有效除了删除明细我还会生成一份“清洗前后指标对比”把核心数字汇总出来。常用的指标包括原始行数、标准化后行数、去重后行数被删除的行数关键字段的唯一值数量比如会员ID去重前多少、去重后多少每个字段的缺失值数量重复率即被删行数占原始行数的比例。这些指标可以整理成一个简单的文本或JSON报告summary { source_file: raw_channels.csv, total_rows_before: len(df_raw), total_rows_after_standardize: len(df_std), total_rows_after_dedup: len(df_kept), removed_rows: len(df_removed), duplicate_rate: round(len(df_removed) / len(df_raw), 4), unique_member_id_before: df_raw[member_id].nunique(), unique_member_id_after: df_kept[member_id].nunique(), } import json with open(output/summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2)这份报告的价值在于你不需要展开细节只看汇总数字就能判断这次清洗是否正常。比如重复率突然从5%飙到40%多半是标准化规则太激进或者subset选错了字段这时就能回到明细里去排查。4.3 文件MD5/SHA256指纹让报告和文件能对上“可检查”的最高级别是给输出的每个文件都生成一个唯一的指纹防止后续文件在传输、下载或转存过程中被无意改动。这就是很多人提到的“CSV文件怎么做MD5校验”。文件级别校验很简单直接对文件计算MD5或SHA256拿Linux/Mac终端来说md5sum output/clean_data.csv sha256sum output/removed_records.csvWindows PowerShell也可以用Get-FileHash -Algorithm SHA256 .\output\clean_data.csv把得到的哈希指纹写进报告文件以后任何人拿到这批CSV都可以重新计算哈希值和报告里的指纹比对确认文件是否被改动过。这有点类似给文件做了个“身份证号”把整个清洗工作的交付状态固化下来。对于行级别的MD5校验之前在第三章也提过就是用哈希指纹给每一行做唯一标记。两种指纹的应用场景不同文件级指纹用于验证交付文件的完整性行级指纹用于行级别的判重与追踪可以搭配使用。5. 一次完整的本地CSV清洗带注释的实战流程前面讲了理论和要点这一章我放一个可以直接照着改的完整示例。假设现在有两份渠道商CSV需要合并、清洗、去重并输出干净文件、删除明细和汇总报告。5.1 场景说明与目录结构先看目录结构project/ ├── raw/ │ ├── channel_a.csv │ └── channel_b.csv ├── output/ │ ├── clean_data.csv │ ├── removed_records.csv │ └── summary.json └── clean_csv.pychannel_a.csv和channel_b.csv字段基本一致但字段格式不完全相同可能有编码差异和重复记录。目标是把两份文件合并成一份干净数据按store_id去重保留最新更新的一条。5.2 分步代码与执行逻辑完整代码如下import pandas as pd import hashlib import json # ---------- 1. 读取与合并 ---------- df_a pd.read_csv(raw/channel_a.csv, dtypestr, encodingutf-8) df_b pd.read_csv(raw/channel_b.csv, dtypestr, encodingutf-8) # 如果列名不一致先统一字段名 df_a.columns [str(c).strip().lower().replace( , _) for c in df_a.columns] df_b.columns [str(c).strip().lower().replace( , _) for c in df_b.columns] df pd.concat([df_a, df_b], ignore_indexTrue) print(f合并后总行数: {len(df)}) # ---------- 2. 标准化 ---------- # 字符串列统一去掉首尾空格和全角空格统一中英文括号 str_cols df.select_dtypes(include[object]).columns for col in str_cols: if col not in [update_time, store_id]: df[col] (df[col] .astype(str) .str.strip() .str.replace(\u3000, , regexFalse) .str.replace(, (, regexFalse) .str.replace(, ), regexFalse)) # 门店ID如果存在非数字字符一律去除 if store_id in df.columns: df[store_id] df[store_id].astype(str).str.replace(r\D, , regexTrue) # 日期字段统一为 YYYY-MM-DD if update_time in df.columns: df[update_time] pd.to_datetime(df[update_time], errorscoerce).dt.strftime(%Y-%m-%d) # 空值占位符统一 df df.replace([, NULL, NA, N/A, None, null, -, nan], pd.NA) df df.fillna() # ---------- 3. 生成行级指纹可选用于快速核对 ---------- def row_fingerprint(row): raw |.join([str(v) for v in row]) return hashlib.md5(raw.encode(utf-8)).hexdigest() df[_row_fingerprint] df.apply(row_fingerprint, axis1) # ---------- 4. 去重先按更新时间排序再按门店ID去重 ---------- df_sorted df.sort_values(update_time, ascendingFalse, na_positionlast) df_kept df_sorted.drop_duplicates(subset[store_id], keepfirst) # 标记所有参与重复的行 df[_dup_flag] df.duplicated(subset[store_id], keepFalse) removed_index df.index.difference(df_kept.index) df_removed df.loc[removed_index].copy() df_removed[_drop_reason] store_id重复保留update_time最新记录 print(f去重后行数: {len(df_kept)}删除行数: {len(df_removed)}) # ---------- 5. 输出文件 ---------- df_kept.drop(columns[_dup_flag]).to_csv(output/clean_data.csv, indexFalse, encodingutf-8-sig) df_removed.drop(columns[_dup_flag]).to_csv(output/removed_records.csv, indexFalse, encodingutf-8-sig) # ---------- 6. 汇总报告 ---------- summary { source_files: [channel_a.csv, channel_b.csv], total_rows_after_merge: len(df), total_rows_after_dedup: len(df_kept), removed_rows: len(df_removed), duplicate_rate: round(len(df_removed) / len(df), 4), unique_store_id_before_dedup: df[store_id].nunique(), unique_store_id_after_dedup: df_kept[store_id].nunique(), } with open(output/summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2) print(清洗完成结果保存在 output/ 目录下)这段代码的逻辑很直白读文件、合并、标准化、打指纹、排序去重、输出明细和报告。你可以根据实际字段名调整subset和排序字段框架不需要动。5.3 跑完之后的检查清单代码能跑通只是第一步我每次清洗完还会做下面这套检查确保结果真的可信看summary.json里的删除行数和重复率如果异常偏高先怀疑标准化阶段是不是把大量有效值误统一成了同一个占位符打开removed_records.csv随机抽几行对照_drop_reason确认删除逻辑正确用文件哈希指纹再核对一遍输出文件的完整性确认没有写入时出现异常如果业务方需要可以把clean_data.csv和removed_records.csv一起交给他们而不是只给一个干净文件。这套检查动作做完我才会放心地把数据交付出去。最后分享一个我自己的习惯清洗脚本本身也要做版本管理每次跑批都在脚本头部写清楚清洗规则和日期输出报告的文件名加上时间戳比如summary_20250411.json。这样一个月后再有人问“这份数据是怎么洗出来的”你翻出当天的脚本、报告和指纹几分钟就能完整还原整个过程。清洗这件事最怕的不是数据脏而是脏完之后没人说得清到底做了什么。把过程留给别人查也留给未来的自己查这才是“可检查的报告”真正的意义。
