数据清洗数据源.zip实战：从脏数据到可复用清洗流水线

简介：这份数据清洗数据源压缩包面向大数据应用学习者与数据分析初学者，提供可直接上手练习的原始数据素材，帮助解决真实数据中缺失值、异常值、重复记录与格式不一致等常见问题。包内共11个文件，涵盖sql、csv、txt、xlsx、xls、json、xml等多种格式，分别对应数据库表结构、结构化表格、文本日志与半结构化数据等典型数据源，压缩包整体约96KB，体积轻便便于快速导入与反复实验。目前已有528人学习下载，说明其在数据清洗入门练习中具有一定参考价值。读者可借助这些素材练习缺失值填充、异常值检测、一致性校验与数据类型转换等操作，并结合Pandas、SQL或ETL工具完成从多源数据提取到清洗落地的完整流程，为后续数据分析与挖掘打下基础。

1. 数据清洗数据源.zip：一个压缩包背后，藏着多少条能跑通的清洗流水线

拿到「数据清洗数据源.zip」这个标题，多数人第一反应是去找下载链接，但真正做过数据工程的人会先问一句：这个包里到底装的是原始脏数据，还是已经配好的清洗脚本加样例数据？两种东西的价值差着数量级。前者是素材，后者是能直接复现的流水线。我见过太多团队把「数据源」理解成「一堆 CSV 扔进文件夹」，结果接手的人打开一看，字段名带空格、日期三种格式混排、空值用NULL、null、-、无四种写法，清洗脚本根本没法通用。这个标题真正指向的需求是：拿到一批有代表性的脏数据样本，配上一套可迁移的清洗逻辑，让新手能照着跑通，让熟手能直接改参数上生产。适合谁？做数据分析前期准备的人、搭 ETL 管道的后端、以及被「数据质量」反复折磨的算法工程师。下面按「先看清数据源长什么样，再动手写清洗，最后避开那些让脚本白写的坑」这条线走。

2. 拆开数据清洗数据源.zip之前：先搞清脏数据的六种典型形态

2.1 为什么不能拿到文件就写dropna()

数据清洗最怕的不是脏，而是不知道脏在哪。一个压缩包解压后，你面对的可能是一张订单表、一份用户行为日志、或者几个爬虫抓下来的 JSON。直接df.dropna()或者df.fillna(0)是最常见的翻车起点——把有意义的缺失当成噪声删掉，或者把「未填写」硬填成 0，后面聚合出来的均值全是错的。我一般会先做一轮「数据体检」，不写任何清洗逻辑，只输出每个字段的：非空率、唯一值数量、前 10 个高频值、以及最值。这一步花十分钟，能省掉后面两小时的返工。

import pandas as pd def data_health_check(df, name="dataset"): """输出每个字段的基础体检报告，不做任何修改""" report = [] for col in df.columns: s = df[col] report.append({ "字段": col, "类型": str(s.dtype), "非空率": round(s.notna().mean(), 4), "唯一值数": s.nunique(dropna=True), "高频值Top3": s.value_counts(dropna=True).head(3).to_dict(), "样例": s.dropna().head(2).tolist() }) return pd.DataFrame(report) # 用法：先读进来，别急着改 df = pd.read_csv("raw_orders.csv", dtype=str, keep_default_na=False) print(data_health_check(df).to_string())

这段代码的关键在dtype=str和keep_default_na=False。前者强制所有列按字符串读，避免 pandas 自动把「00123」变成 123、把「2024-01-01」猜成 datetime 后又在后续操作里丢精度；后者让空字符串保持空字符串，而不是被自动转成NaN，这样你才能区分「字段缺失」和「字段填了空」。体检报告里如果某列非空率低于 60%，先别删，去问数据来源方这个字段是不是选填；如果唯一值数等于行数，大概率是主键或 ID，清洗时要保护。

2.2 六种脏数据形态与对应识别信号

把常见脏数据归成六类，每类都有可观测的信号，不用靠猜。

识别信号比清洗动作更重要。比如「格式不一致」这一条，如果你只看到日期列有2024/1/5和2024-01-05两种，直接pd.to_datetime可能报错，但更隐蔽的是05/01/2024这种日月顺序歧义——美式还是欧式？这时候不能靠代码猜，要去问数据提供方，或者看同一行其他字段的上下文。我一般会在体检报告里加一列「格式模式」，用正则粗提取，把明显不同的模式标出来。

2.3 从压缩包到 DataFrame：读取阶段的三个参数决定后面顺不顺

解压后常见格式是 CSV、Excel、JSON、Parquet。读取阶段有三个参数最容易被忽略，但直接影响清洗难度。

# CSV：指定编码和分隔符，别让 pandas 猜 df_csv = pd.read_csv( "raw_data.csv", encoding="utf-8-sig", # 处理带 BOM 的 UTF-8，避免第一列名带 \ufeff sep=",", # 如果实际是 \t 或 |，这里必须改 dtype=str, # 统一按字符串读，后续按需转换 keep_default_na=False, # 保留空字符串，区分缺失与空值 on_bad_lines="warn" # 遇到列数不对的行先警告，不直接崩 ) # Excel：指定 sheet 和表头行，避免把标题行当数据 df_excel = pd.read_excel( "raw_data.xlsx", sheet_name="Sheet1", header=0, # 如果前两行是说明，改成 header=2 dtype=str ) # JSON：如果是嵌套结构，先 json_normalize 再进 DataFrame import json with open("raw_data.json", "r", encoding="utf-8") as f: raw = json.load(f) df_json = pd.json_normalize(raw, sep="_") # 嵌套字段用下划线连接

encoding="utf-8-sig"是血泪经验：Windows 导出的 CSV 经常带 BOM，用utf-8读第一列名会多一个不可见字符，后面df["订单号"]死活匹配不上。on_bad_lines="warn"在数据量大时很有用，先跑通再回头处理坏行，不要一上来就error中断。JSON 的json_normalize把嵌套字典展开成扁平列，sep="_"让user.address.city变成user_address_city，后续引用方便。

3. 清洗流水线的四步骨架：从原始表到可用表

3.1 第一步：列名与类型标准化，别让中文空格毁掉 merge

列名标准化是清洗的第一道工序，也是最容易被跳过的一步。常见问题：列名带前后空格、大小写混用、中文列名里夹英文括号、重复列名 pandas 自动加.1。这些在单表操作时没事，一旦要merge或写 SQL，就是灾难。

import re def standardize_columns(df): """列名标准化：去空格、转小写、替换特殊字符、去重""" new_cols = [] seen = {} for col in df.columns: c = str(col).strip().lower() c = re.sub(r"[\s\-/\\]+", "_", c) # 空格和斜杠转下划线 c = re.sub(r"[^\w\u4e00-\u9fa5]", "", c) # 保留字母数字下划线和中文 if c in seen: seen[c] += 1 c = f"{c}_{seen[c]}" else: seen[c] = 0 new_cols.append(c) df.columns = new_cols return df df = standardize_columns(df)

逻辑说明：先strip去首尾空格，再lower统一大小写，然后把空格、斜杠、反斜杠替换成下划线，接着用正则去掉除字母、数字、下划线、中文以外的字符。最后用seen字典处理重名，第二个同名列变成_1后缀。参数上，\u4e00-\u9fa5是中文 Unicode 范围，如果你的数据有日文或韩文列名，需要扩展这个范围。这一步做完，后面所有引用都用标准化后的列名，不再出现df[" 订单号 "]这种写法。

类型标准化紧随其后。字符串列统一astype(str)并strip；数值列先用pd.to_numeric(errors="coerce")转，转失败的置为NaN后续单独处理；日期列用pd.to_datetime(errors="coerce", format="mixed")，format="mixed"让 pandas 逐行推断格式，比统一格式更宽容，但速度慢，数据量大时建议先抽样确定格式再指定。

3.2 第二步：缺失值与重复值的处理策略，按字段语义分治

缺失值处理没有万能公式，核心原则是「按字段语义分治」。我一般把字段分成三类：标识类（ID、主键）、维度类（地区、品类）、度量类（金额、数量）。标识类缺失直接标记为无效行；维度类缺失填「未知」并保留，因为「未知」本身可能是信息；度量类缺失要看业务，能回算的回算，不能回算的标记为缺失，不要填 0。

def handle_missing(df, id_cols, dim_cols, measure_cols): """按字段语义处理缺失值""" # 标识类缺失：整行标记，不删除，留待人工确认 df["_invalid"] = df[id_cols].isna().any(axis=1) # 维度类缺失：填「未知」 for col in dim_cols: df[col] = df[col].fillna("未知") # 度量类缺失：保留 NaN，额外加标记列 for col in measure_cols: df[f"{col}_missing"] = df[col].isna() return df df = handle_missing( df, id_cols=["order_id", "user_id"], dim_cols=["province", "category"], measure_cols=["amount", "quantity"] )

重复值处理要区分「完全重复」和「业务重复」。完全重复用df.drop_duplicates()直接删；业务重复比如同一用户同一秒提交两次订单，需要按业务规则保留最新或最早一条。我一般先看重复组的数量，如果重复率超过 5%，先别删，去查数据来源是不是有重放或多次导入。

# 完全重复 before = len(df) df = df.drop_duplicates() print(f"删除完全重复: {before - len(df)} 行") # 业务重复：同一 order_id 保留更新时间最新的一条 df = df.sort_values("update_time").drop_duplicates( subset=["order_id"], keep="last" )

keep="last"配合先按时间排序，保留最新记录。如果时间列也有缺失，排序会把 NaN 放最后，需要先处理时间缺失再排序。

3.3 第三步：异常值检测与逻辑校验，用 IQR 和业务规则双保险

异常值检测分统计法和业务法。统计法常用 IQR（四分位距），对偏态分布比 3σ 稳健；业务法靠硬规则，比如年龄 0-120、金额非负、结束时间不早于开始时间。两者结合，统计法发现可疑点，业务法做最终判定。

def detect_outliers_iqr(series, factor=1.5): """IQR 法标记异常值，返回布尔 Series""" q1 = series.quantile(0.25) q3 = series.quantile(0.75) iqr = q3 - q1 lower = q1 - factor * iqr upper = q3 + factor * iqr return (series < lower) | (series > upper) # 对金额列做 IQR 检测 df["amount_outlier"] = detect_outliers_iqr(df["amount"].astype(float)) # 业务规则校验 df["logic_error"] = ( (df["amount"].astype(float) < 0) | (df["end_time"] < df["start_time"]) )

factor=1.5是标准 IQR 系数，调大到 3 会减少标记数量，适合数据本身波动大的场景。amount_outlier和logic_error都只做标记，不直接删或改，因为异常值可能是真实的高价值订单，也可能是录入错误，需要人工或下游规则判断。这一步的输出是一张带标记列的表，而不是一张「干净」的表——保留标记比直接清洗更安全，后悔药就在这些标记列里。

3.4 第四步：输出与版本记录，让清洗过程可回溯

清洗完不记录版本，等于白洗。我一般输出三样东西：清洗后的主表、清洗日志（每步删了多少行、改了多少值）、以及字段字典（每个字段的含义、类型、取值范围）。清洗日志用 JSON 存，方便后续对比不同批次。

import json from datetime import datetime clean_log = { "source_file": "raw_orders.csv", "clean_time": datetime.now().isoformat(), "steps": [ {"step": "standardize_columns", "rows": len(df)}, {"step": "handle_missing", "invalid_rows": int(df["_invalid"].sum())}, {"step": "drop_duplicates", "removed": before - len(df)}, {"step": "detect_outliers", "amount_outliers": int(df["amount_outlier"].sum())} ], "output_rows": len(df) } with open("clean_log.json", "w", encoding="utf-8") as f: json.dump(clean_log, f, ensure_ascii=False, indent=2) df.to_parquet("clean_orders.parquet", index=False)

输出用 Parquet 而不是 CSV，因为 Parquet 保留类型信息，下次读取不用再猜编码和分隔符，体积也更小。clean_log.json里记录每一步的行数变化，下次数据更新时对比日志，能快速定位是哪一步导致行数异常波动。

4. 避坑与排查：数据清洗数据源.zip落地时最容易翻车的五件事

4.1 现象：pd.to_datetime报错OutOfBoundsDatetime；原因：日期超出 pandas 时间戳范围；解决：先筛出异常年份再转换

pandas 的Timestamp默认范围是 1677 到 2262 年。如果数据里有9999-12-31这种「永久有效」的占位日期，pd.to_datetime会直接抛OutOfBoundsDatetime。解决方法是先用字符串截取年份，把明显超出范围的筛出来单独处理。

# 先提取年份，筛出可疑行 df["year_str"] = df["date_col"].astype(str).str[:4] suspicious = df[~df["year_str"].str.match(r"^(19|20)\d{2}$")] print(f"可疑日期行数: {len(suspicious)}") # 对正常范围的行做转换 df["date_parsed"] = pd.to_datetime( df.loc[~df.index.isin(suspicious.index), "date_col"], errors="coerce" )

4.2 现象：merge后行数暴涨；原因：关联键有重复值；解决：merge 前先检查关联键唯一性

两个表merge后行数从 1 万变成 5 万，八成是右表的关联键不唯一。merge默认是内连接，一对多会展开成多行。解决：merge 前对关联键做value_counts，确认唯一再合并。

# 检查关联键唯一性 right_dup = df_right["key"].duplicated().sum() if right_dup > 0: print(f"右表关联键有 {right_dup} 个重复，merge 会膨胀") # 先聚合或去重 df_right = df_right.drop_duplicates(subset=["key"], keep="first") df_merged = df_left.merge(df_right, on="key", how="left", validate="one_to_one")

validate="one_to_one"让 pandas 在键不唯一时直接报错，比事后查行数更早发现问题。

4.3 现象：中文列名在groupby后消失；原因：列名含不可见字符；解决：统一用standardize_columns处理

从 Excel 复制过来的列名经常带\xa0（不换行空格）或\u200b（零宽空格），肉眼看不出来，但df["订单号"]就是匹配不上。standardize_columns里的正则[^\w\u4e00-\u9fa5]会把这些不可见字符去掉。如果已经进了流程，可以用df.columns = df.columns.str.replace(r"\s+", "", regex=True)应急。

4.4 现象：fillna(0)后均值失真；原因：缺失被当成真实 0；解决：度量类字段保留 NaN，用标记列区分

金额、数量、时长这类度量字段，缺失和 0 是两回事。缺失代表「没记录」，0 代表「记录为 0」。填 0 后算平均，分母变大，均值被拉低。正确做法是保留 NaN，df["amount"].mean()会自动跳过 NaN，同时加一列amount_missing标记缺失位置，后续分析可以按缺失与否分组对比。

4.5 现象：清洗脚本在样本上跑通，全量数据报内存错误；原因：dtype=str读全量后内存翻倍；解决：分块读取或指定列类型

dtype=str在样本上没问题，但全量数据读进来后，字符串列的内存占用远高于数值列。如果数据超过内存一半，就会 OOM。解决：用chunksize分块处理，或者读之前先指定每列类型，数值列用float32/int32而不是默认的float64/int64。

# 分块读取，每块 10 万行 chunk_list = [] for chunk in pd.read_csv("big_file.csv", chunksize=100000, dtype=str): chunk = standardize_columns(chunk) chunk_list.append(chunk) df = pd.concat(chunk_list, ignore_index=True)

分块时注意standardize_columns要在每块上单独做，因为列名标准化不依赖跨块信息。如果清洗逻辑涉及全局统计（比如 IQR 的上下界），需要先扫一遍算全局分位数，再分块应用。

5. 把清洗脚本变成可复用资产：参数化与自动化校验的两个技巧

清洗脚本写完不是终点，能复用到下一批数据才是。我一般做两件事：把硬编码的列名、阈值、文件路径抽成配置字典；在脚本末尾加一段自动化校验，用断言检查清洗后的表是否满足基本约束。

# config.yaml 或 config.py CONFIG = { "id_cols": ["order_id", "user_id"], "dim_cols": ["province", "category"], "measure_cols": ["amount", "quantity"], "iqr_factor": 1.5, "date_col": "create_time", "output_path": "clean_orders.parquet" } # 自动化校验 def validate_clean(df, config): """清洗后校验，不通过直接抛异常""" assert df[config["id_cols"]].notna().all().all(), "标识列仍有缺失" assert df["amount"].dropna().ge(0).all(), "金额出现负值" assert df["create_time"].dropna().between( "2020-01-01", "2030-01-01" ).all(), "日期超出合理范围" print("校验通过") validate_clean(df, CONFIG)

配置化的好处是换一批数据只改CONFIG，不动清洗逻辑。自动化校验用assert把「隐性假设」变成「显性检查」——比如你默认金额非负，那就写出来，下次数据里出现负金额时脚本直接报错，而不是悄悄算出一个负的总额。校验规则不用多，覆盖标识列、关键度量、时间范围这三类就够。

另一个技巧是给清洗脚本加--dry-run参数，只输出日志不写文件。这样在正式跑之前可以先看一遍每步删了多少行、标记了多少异常，确认无误再落盘。我自己的习惯是：任何清洗脚本第一次跑全量数据，必须--dry-run过一遍，看日志里的行数变化是否符合预期，再正式输出。这个习惯帮我拦下过好几次「关联键重复导致行数膨胀」的事故。

数据清洗这件事，工具和代码都是次要的，核心是「先看清再动手，保留标记不硬删，日志留全可回溯」。希望帮到你。

本文还有配套的精品资源，点击获取