主数据清洗实战：Python分层流水线与20个高可用库

1. 项目概述：为什么“主数据清洗”不是可选项，而是数据工作的生死线

在真实的数据项目里，我见过太多团队把80%的时间花在写模型、调参数、画图表上，结果上线后发现——输入数据里混着2019年的客户地址、重复的SKU编码、用中文顿号分隔的JSON字段、还有Excel里那个永远删不干净的“合计”行。这时候再强的算法也救不了。所谓“主数据”（Master Data），说白了就是企业最核心、最权威、最需要跨系统共享的那一套基础信息：客户是谁、产品是什么、供应商在哪、组织架构怎么搭。它不是某张表里的几列数据，而是整个业务系统的“DNA”。而“Wrangling”这个词，比“Cleaning”更狠——它不是擦灰，是驯服。是把散落在CRM、ERP、Excel、邮件附件、甚至纸质扫描件里的原始数据，像驯马一样拉进统一围栏，剪毛、打钉、配鞍、训练，直到能听指令、跑得稳、不尥蹶子。标题里这个“First”，不是时间顺序，是优先级排序：没搞定主数据清洗，后面所有分析、建模、BI看板，都是在流沙上盖楼。我带过的三个金融风控项目，最终上线延迟超过40天，全卡在客户主数据的身份证号格式校验、手机号脱敏一致性、以及历史开户日期与工商注册日期的逻辑对齐上。这20个Python库，不是工具清单，而是20把不同形状的扳手；那15条实践，也不是教条，而是我在凌晨三点对着Jupyter Notebook里报错的ValueError: cannot convert float NaN to integer反复重试后，用咖啡和黑眼圈换来的肌肉记忆。

2. 核心思路拆解：为什么不用Pandas“一把梭”，而要构建分层清洗流水线

很多人一上来就打开pandas，df.dropna()、df.fillna()、df.astype()三连击，觉得数据“看起来干净了”。但主数据清洗不是美颜滤镜，它是外科手术。我见过最典型的失败案例，是一家零售企业用df.fillna(method='ffill')填充缺失的门店区域代码，结果把华东区的店全填成了华北区的编码，导致季度销售归因完全错位。问题出在哪儿？在于混淆了“缺失值处理”和“业务规则注入”。主数据清洗必须分层推进，每一层解决一类问题，且层与层之间有明确的依赖和验证点。我们团队现在强制执行四层流水线：探查层 → 标准化层 → 关联层 → 治理层。

• 探查层（Discovery Layer）：目标不是改数据，是读懂数据。用pandera定义Schema断言，用ydata-profiling生成交互式报告，重点看“为什么缺失”——是系统未采集（真缺失），还是录入员偷懒填了“暂无”（伪缺失），或是上游接口传了空字符串（格式污染）。这一层输出的不是清洗后的表，而是一份《数据健康白皮书》，包含每个字段的唯一值分布、异常值阈值、跨字段逻辑矛盾点（比如“客户等级为VIP”但“累计消费额为0”）。没有这份白皮书，后面的清洗就是蒙眼开车。

• 标准化层（Standardization Layer）：这才是传统意义的“清洗”，但必须绑定业务字典。比如处理地址，address_parser库能识别“上海市浦东新区张江路123号”，但无法判断“张江路”该归属“浦东新区”还是“张江镇”。这时必须接入企业内部的《行政区划标准码表》，用fuzzywuzzy做模糊匹配，再用geopy做地理坐标反查二次校验。标准化不是让数据变“整齐”，是让数据变“可解释”。

• 关联层（Linkage Layer）：主数据的灵魂在于“唯一性”。一个客户在CRM叫“张三”，在ERP叫“张三丰”，在财务系统叫“ZS001”，这三条记录必须合并为一条主记录。recordlinkage和dedupe是主力，但关键在Blocking策略——不能让算法去两两比对100万条记录（O(n²)复杂度会爆炸）。我们通常先用邮政编码+姓名拼音首字母做粗筛，再用电话号码哈希值做精筛，最后用jaro_winkler计算姓名相似度。实测下来，Blocking后候选对从10亿对降到200万对，清洗耗时从3天压缩到47分钟。

• 治理层（Governance Layer）：清洗完不是终点，是起点。用great_expectations配置数据质量检查点，比如“客户主键重复率<0.001%”、“手机号格式合规率>99.95%”，这些检查必须嵌入ETL调度任务，失败则自动告警并阻断下游流程。治理层让清洗动作可审计、可回滚、可度量。

这套分层思路的核心逻辑很朴素：把“人”的业务知识，翻译成“机器”能执行的、可验证的规则链。Pandas是刀，但刀法得按解剖图谱来，不能乱砍。

3. 20个核心Python库深度解析：不是罗列，是按实战场景配装

市面上的Python数据清洗库动辄上百，但真正能在主数据项目中扛住压力、解决具体痛点的，也就这20个。我按它们在四层流水线中的核心作用分类，并标注每个库在真实生产环境中的“生存率”（基于我经手的17个项目统计：稳定可用率>95%为★，80%-95%为★★，<80%为★★★需谨慎）。

3.1 探查层：读懂数据的“CT机”

• pandera 0.15+（★）：不是简单的类型校验，而是用声明式语法定义业务约束。比如客户表要求：“customer_id必须是8位数字字符串，registration_date不能晚于当前日期，age必须在0-120之间，且当customer_type为'CORPORATE'时，tax_id不能为空”。代码写成：

  import pandera as pa from pandera import Check, Column, DataFrameSchema customer_schema = DataFrameSchema({ "customer_id": Column(pa.String, checks=Check.str_matches(r"^\d{8}$")), "registration_date": Column(pa.DateTime, checks=Check.less_than_or_equal_to(pd.Timestamp.now())), "age": Column(pa.Int, checks=[Check.greater_than_or_equal_to(0), Check.less_than_or_equal_to(120)]), "customer_type": Column(pa.String), "tax_id": Column(pa.String, nullable=True), }, # 关键：跨字段条件检查 checks=Check( lambda df: df[df["customer_type"] == "CORPORATE"]["tax_id"].notna().all(), name="corporate_tax_id_required" ))

  提示：pandera的validate方法返回带详细错误路径的SchemaError，比assert报错信息精准10倍，调试时直接定位到第1234行第5列。

• ydata-profiling 4.6+（★）：替代df.describe()的终极方案。生成HTML报告包含：字段相关性热力图（快速发现“客户等级”和“年消费额”强正相关）、缺失值矩阵（看出缺失是否集中于某几个系统来源）、文本字段的词频TOP20（发现“北京市”被录成“北京”“BJ”“京市”等12种变体）。我们把它集成进Airflow DAG，每次ETL跑完自动生成报告存入S3，BI团队直接看链接，不再找数据工程师要“数据情况说明”。

• missingno 0.5+（★★）：专治“缺失模式”。msno.matrix(df)可视化缺失值分布，能一眼看出：缺失是否随机（散点状），还是按时间批次集中（竖条状），或是按业务线成块（方块状）。曾靠这个图发现某渠道API在每月5号凌晨2点固定超时，导致当月前5天客户注册数据全空——这是df.isnull().sum()永远发现不了的系统性故障。

3.2 标准化层：让数据“说同一种话”

• address_parser 0.2+（★★）：轻量级地址解析，适合国内简单场景。但注意：它把“广东省深圳市南山区科技园科苑路15号”拆成{"province": "广东省", "city": "深圳市", ...}，却无法识别“科苑路15号”属于“腾讯大厦”还是“大疆总部”。所以必须配合企业自有POI库做二次映射。我们用redis缓存POI关键词，解析后查缓存，命中率提升至92%。

• fuzzywuzzy 0.18+（★）：字符串模糊匹配的基石。但默认的fuzz.ratio()对长文本效果差，主数据常用的是fuzz.token_sort_ratio()（忽略词序）和fuzz.WRatio()（加权组合多种算法）。处理供应商名称时，用process.extractOne("深证市腾迅科技有限公司", supplier_list, scorer=fuzz.token_sort_ratio)比单纯ratio准确率高37%。实测技巧：预处理时统一转小写、去标点、替换“有限公司”为“公司”，再匹配，效果翻倍。

• phonenumbers 8.13+（★）：处理全球手机号的黄金标准。不仅能校验“13812345678”是否为中国大陆有效号段，还能提取国家码、运营商、甚至判断是否虚拟号段（如阿里小号）。关键参数：region='CN'指定中国区，carrier模块可查号段归属。曾用它揪出一批用“170/171”虚拟号段注册的羊毛党账号，拦截率99.4%。

• dateutil 2.8+（★）：parser.parse()能识别“2023-03-15”、“15/03/2023”、“2023年3月15日”等20+种格式，但必须设default=datetime(1970,1,1)防歧义。主数据中常见陷阱是“2023/03/15”被误读为“2023年15月3日”，dateutil会抛ValueError，而pandas.to_datetime(errors='coerce')直接变成NaT——后者更危险，因为缺失值会被后续流程静默忽略。

3.3 关联层：给数据“做亲子鉴定”

• recordlinkage 3.0+（★）：专为主数据实体解析设计。核心是Index类构建候选对，Compare类定义比较规则，Classify类做最终判定。我们最常用的Blocking策略是Block('postal_code')+Block('first_name', 'last_name')，再用Compare().string('email', method='jarowinkler', threshold=0.85)。关键经验：threshold不能拍脑袋定，要用已知的1000对真实匹配/不匹配样本做ROC曲线，选F1-score最高点。

• dedupe 2.1+（★★★）：基于机器学习的关联，适合无先验规则的场景。但训练成本高——需要人工标注300+对样本，且模型对新数据泛化能力弱。我们只在历史数据迁移项目中用它，日常ETL坚决不用。它的优势是能发现recordlinkage漏掉的“张三”和“张叁”这种字形变异。

• rapidfuzz 3.0+（★）：fuzzywuzzy的高性能替代品，Cython加速，速度提升5-8倍。主数据清洗常需对百万级客户名做批量模糊匹配，rapidfuzz.process.cdist()比fuzzywuzzy.process.extractOne快得多。注意：API几乎兼容，只需import rapidfuzz as fuzz即可平滑替换。

3.4 治理层：让清洗“可审计、可追溯”

• great_expectations 0.17+（★）：不是校验工具，是数据契约框架。expect_column_values_to_match_regex检查手机号，expect_table_row_count_to_be_between防数据截断，expect_column_pair_values_A_to_be_greater_than_B确保“结束日期>=开始日期”。所有Expectation存为JSON，Git版本管理，CI/CD自动执行。某次上线前，great_expectations发现测试环境客户表比生产少23条，追查发现是上游SQL漏写了WHERE status='ACTIVE'——避免了一次重大数据事故。

• sdv 1.10+（★★）：合成数据生成，用于测试。主数据清洗脚本不能拿真实客户数据练手，sdv能生成符合原始分布的假数据：Synthesizer保持字段相关性，“张三”的年龄、职业、城市消费水平都符合真实规律。我们用它生成10万条测试数据，覆盖所有边界case（如0岁婴儿、120岁老人、手机号含字母等），清洗脚本通过率从82%提升到100%。

• polars 0.19+（★）：不是替代pandas，是处理超大主数据集的“涡轮增压器”。当客户主表突破5000万行，pandas内存暴涨、GC卡顿，polars用Rust写的引擎，内存占用降60%，groupby().agg()快3倍。关键适配：polars的lazy API必须显式.collect()才执行，避免意外触发计算。

注意：所有库的版本号都标注了，因为主数据项目对稳定性要求极高。pandera 0.14的Check语法和0.15不兼容，升级前必须全量回归测试。我们团队的《Python库选型清单》里，明确写着：“非必要不升级，升级必测全量主数据表”。

4. 15条主数据清洗最佳实践：来自血泪教训的硬核笔记

这15条不是教科书摘抄，是我在12个主数据项目里，踩过坑、修过bug、熬过夜后，用Markdown记在Notion里的“防坑指南”。每一条都对应一个真实故障场景。

4.1 数据探查阶段：别急着写代码，先问三个问题

1. “这个字段的‘空’，业务上代表什么？”
   曾处理一个医疗客户主数据，emergency_contact_phone字段大量为空。开发直接fillna('未知')，结果上线后客服接到投诉：“为什么我的紧急联系人变成‘未知’了？！” 实际业务含义是“客户未提供”，应填None或留空，而非字符串。正确做法：查《业务字典》，确认空值语义，再决定是drop、fill还是标记为NULL_REASON新字段。

2. “缺失是随机的，还是有模式的？”
   用missingno看缺失矩阵，发现customer_level字段在2022年Q3后突然缺失率飙升至40%。追查发现是CRM系统升级，新字段vip_tier替代了旧字段，但ETL脚本没更新。模式化缺失往往指向系统变更，不是数据问题。

3. “这个值看起来奇怪，是脏数据，还是新业务规则？”
   客户表里出现country_code='XX'，pandas报错'XX' not in country_dict。我以为是脏数据，准备drop。结果业务方说：“这是新上线的‘国际业务部’专属代码，下周正式启用”。主数据清洗必须和业务方建立“异常值日报”机制，每天同步TOP10异常值，共同决策。

4.2 标准化阶段：规则即法律，但法律要有人情味

4. “永远不要用df.fillna(method='ffill')处理主键字段”
   血泪教训：某次修复客户ID缺失，用ffill填充，结果把A客户的ID填给了B客户，导致B客户的订单全算到A名下。主键缺失必须drop或generate new ID，绝不能用邻近值填充。

5. “日期标准化，必须锁定时区”
   pd.to_datetime('2023-01-01')默认UTC，但中国业务数据是Asia/Shanghai。to_datetime(..., utc=True).dt.tz_convert('Asia/Shanghai')才是安全写法。否则“2023-01-01 00:00:00 UTC”会变成“2023-01-01 08:00:00 CST”，跨日分析全错。

6. “手机号清洗，先校验再格式化”
   错误做法：df['phone'].str.replace(r'\D', '')直接删所有非数字。结果把“138-1234-5678”变成“13812345678”，但把“+86 138 1234 5678”也变成“8613812345678”（多出86）。正确流程：phonenumbers.parse()校验合法性 →format_number(..., PhoneNumberFormat.E164)标准化为+8613812345678→ 再按需截取纯数字。

7. “地址标准化，宁可保守，不可激进”
   address_parser可能把“上海徐汇区漕溪北路123号”错分为{"city":"上海","district":"徐汇区","road":"漕溪北路","number":"123号"}，但实际“漕溪北路”是道路名，“123号”是门牌号。若强行用district字段做区域分析，会把漕河泾开发区的企业全算进徐汇区。我们的规则：address_parser结果仅作参考，最终以高德/百度地图API地理编码返回的adcode（行政区划代码）为准。

4.3 关联阶段：合并不是目的，唯一性才是

8. “Blocking策略，必须用业务维度，而非技术维度”
   错误：用hash(customer_name)做Blocking，结果“张三”和“张叁”哈希值不同，永远进不了候选对。正确：用soundex(first_name)（发音编码）+levenshtein(last_name, 2)（编辑距离≤2）组合，覆盖同音不同字。

9. “匹配阈值，必须用业务样本校准，而非经验值”
   fuzz.ratio()阈值设0.8？某次用0.8匹配客户名，漏掉了“北京字节跳动科技有限公司”和“北京字节跳动网络技术有限公司”（ratio=0.79）。后来用1000对已知匹配样本画ROC曲线，发现0.75时F1最高。阈值是业务指标，不是技术参数。

10. “合并后，必须保留原始来源和置信度”
    合并“张三”（CRM）和“张三丰”（ERP）为一条主记录，不能简单丢弃原始数据。必须新增字段：source_systems=['CRM','ERP']、merge_confidence=0.92、merged_at='2023-10-01'。这是主数据可追溯性的生命线，审计时全靠它。

4.4 治理与交付阶段：清洗完成，责任才开始

11. “清洗脚本，必须自带‘干运行’模式”
    --dry-run参数：只输出将要修改的行数、字段、值，不真正写入。上线前必跑，让业务方确认“预计修正127个手机号，其中5个将从‘1385678’变为‘+861385678’”。避免“一键清洗”变“一键灾难”。

12. “主数据表，必须有‘最后清洗时间’和‘清洗版本号’字段”
    last_wrangled_at: datetime和wrangle_version: str（如v2.3.1）。下游系统读取数据时，可判断“这个客户信息是上周清洗的，版本v2.1，而最新版v2.3已修复地址格式问题”，决定是否刷新缓存。

13. “清洗日志，必须结构化，且留存≥180天”
    用structlog记录：{"event":"field_cleaned", "field":"phone", "rows_affected":127, "old_value_pattern":"\d{11}", "new_value_pattern":"+86\d{11}"}。JSON格式，方便ELK搜索。曾靠日志快速定位：某天凌晨清洗脚本把所有手机号前加了+86，但漏了判断原格式，导致+86+86138...的错误。

14. “交付物，不是清洗后的CSV，而是‘数据契约’文档”
    包含：Schema定义（pandera代码）、清洗规则说明（如“手机号统一E164格式”）、异常处理策略（如“地址解析失败，保留原文，标记address_parse_status='FAILED'”）、质量指标（“手机号合规率99.98%”）。这是数据团队和业务方的“合同”，签字确认。

15. “永远假设下游会误用你的数据”
    在清洗后脚本末尾，强制添加：

    # 防呆检查：确保关键字段无空值 assert df['customer_id'].notna().all(), "ERROR: customer_id has null values after wrangling!" assert (df['phone'].str.match(r'^\+86\d{11}$').fillna(False) | df['phone'].isna()).all(), "ERROR: phone format invalid"

    这些assert是最后一道保险，比任何文档都管用。它不会让数据变好，但能让问题在最早时间暴露。

5. 实操全流程演示：从一份混乱的客户Excel到可交付主数据表

现在，用一个真实案例，把前面所有理念串起来。假设你收到一份名为raw_customer_q3_2023.xlsx的文件，来自销售同事的本地整理，共8721行。目标：产出符合企业主数据标准的customer_master_v202310.parquet。

5.1 第一步：探查——用10分钟看清数据底细

# 安装探查工具 pip install ydata-profiling missingno pandera

import pandas as pd import ydata_profiling as yp from ydata_profiling import ProfileReport import missingno as msno import matplotlib.pyplot as plt # 1. 加载原始数据（注意：不跳过任何行，保留所有“脏”） df = pd.read_excel("raw_customer_q3_2023.xlsx", keep_default_na=False, na_values=['', 'N/A', 'NULL']) # 2. 生成探查报告（关键：设置minimal=False，获取全部细节） profile = ProfileReport(df, title="Customer Raw Data Health Report", minimal=False) profile.to_file("customer_raw_report.html") # 打开HTML，重点看： # - Overview页的Missing Values矩阵（看缺失模式） # - Variables页每个字段的Distribution（看异常值，如age=999） # - Correlations页的字段相关性（看customer_level和annual_revenue是否强相关） # 3. 可视化缺失模式 plt.figure(figsize=(12, 6)) msno.matrix(df) plt.title("Missing Values Pattern") plt.savefig("missing_matrix.png") # 4. 快速统计关键问题 print("=== Quick Stats ===") print(f"Total rows: {len(df)}") print(f"Rows with any null: {df.isnull().any(axis=1).sum()}") print(f"Unique customer_id count: {df['customer_id'].nunique()} vs total: {len(df)}") # 发现重复！ print(f"Phone field unique values: {df['phone'].nunique()}") # 看有多少种格式

探查发现（真实结果）：

• 缺失矩阵显示：address和postal_code缺失高度重合（同一行都空），registration_date在2023-07-01后全空（系统切换期）。
• customer_id有8721行，但nunique=8715，存在6个重复ID。
• phone字段有127种不同格式：13812345678、+86-138-1234-5678、138 1234 5678、手机：13812345678。
• age字段最大值为199（明显录入错误）。

实操心得：探查阶段绝不写清洗代码！只记录问题。我习惯用Excel另存一份findings_summary.xlsx，每行一个问题，标注严重等级（P0-P2）和业务影响。这份文档，是后续所有工作的起点。

5.2 第二步：标准化——逐字段攻坚，规则驱动

基于探查结果，编写标准化脚本standardize_customer.py：

import pandas as pd import re from phonenumbers import parse, format_number, PhoneNumberFormat, NumberParseException import warnings warnings.filterwarnings("ignore") # phonenumbers会警告，忽略 def clean_phone(phone): """手机号标准化：校验 -> E164 -> 纯数字""" if pd.isna(phone) or not str(phone).strip(): return None try: # 移除所有非数字字符，但保留+号 cleaned = re.sub(r'[^\d+]', '', str(phone)) # 如果以+开头，尝试解析 if cleaned.startswith('+'): num_obj = parse(cleaned, None) else: # 无+号，假设为中国大陆号 num_obj = parse(cleaned, 'CN') # 格式化为E164 e164 = format_number(num_obj, PhoneNumberFormat.E164) # 提取纯数字（去掉+） digits_only = e164[1:] if e164.startswith('+') else e164 return digits_only except NumberParseException: return None # 解析失败，置空 def clean_age(age): """年龄清洗：范围校验 + 异常值处理""" if pd.isna(age): return None try: age_int = int(float(age)) # 处理"25.0"或"25"字符串 if 0 <= age_int <= 120: return age_int else: return None # 199等异常值置空 except (ValueError, TypeError): return None # 主清洗流程 df = pd.read_excel("raw_customer_q3_2023.xlsx", keep_default_na=False, na_values=['', 'N/A', 'NULL']) # 1. 去重：保留第一次出现的记录（按业务规则，首次录入为准） df = df.drop_duplicates(subset=['customer_id'], keep='first') # 2. 清洗手机号 df['phone_clean'] = df['phone'].apply(clean_phone) # 3. 清洗年龄 df['age_clean'] = df['age'].apply(clean_age) # 4. 地址标准化（简化版，实际用address_parser+API） df['address_clean'] = df['address'].str.replace(r'[^\w\u4e00-\u9fff\s\-]', '', regex=True).str.strip() # 5. 日期处理：用dateutil安全解析 from dateutil import parser def safe_parse_date(date_str): if pd.isna(date_str) or not str(date_str).strip(): return None try: # 尝试解析，设默认日期为2023-01-01（避免1970-01-01） dt = parser.parse(str(date_str), default=pd.Timestamp('2023-01-01')) # 确保在合理范围内 if pd.Timestamp('1900-01-01') < dt < pd.Timestamp('2100-01-01'): return dt else: return None except: return None df['registration_date_clean'] = df['registration_date'].apply(safe_parse_date) # 6. 保存中间结果（供业务方确认） df.to_parquet("customer_standardized_intermediate.parquet", index=False) print("Standardization done. Intermediate file saved.")

关键参数选择说明：

• clean_phone中parse(cleaned, 'CN')：强制指定中国区，避免parse('13812345678')因无上下文误判为美国号。
• age_clean中int(float(age))：处理Excel里存储为“25.0”的浮点数，int("25.0")会报错，int(float("25.0"))安全。
• safe_parse_date中default=pd.Timestamp('2023-01-01')：防止parser.parse("2023-03")默认补全为2023-03-01 00:00:00，但业务上“2023-03”可能指整月，应保留为字符串或标记为PARTIAL_DATE。

5.3 第三步：关联——合并CRM与ERP的客户记录

假设我们已有另一份ERP导出的客户表erp_customer.csv，需与清洗后的raw_customer_q3_2023合并。

import recordlinkage import pandas as pd # 加载两份数据 df_raw = pd.read_parquet("customer_standardized_intermediate.parquet") df_erp = pd.read_csv("erp_customer.csv") # 1. 构建索引（Blocking） indexer = recordlinkage.Index() # 用邮政编码和姓名首字母粗筛，大幅减少候选对 indexer.block(left_on='postal_code', right_on='zip_code') indexer.block(left_on='customer_name', right_on='name', method='qgram', q=2) # 2-gram分词 candidate_links = indexer.index(df_raw, df_erp) # 2. 字符串比较 compare_cl = recordlinkage.Compare() compare_cl.string('customer_name', 'name', method='jarowinkler', threshold=0.85, label='name_sim') compare_cl.string('phone_clean', 'mobile', method='exact', label='phone_exact') # 手机号精确匹配权重最高 compare_cl.exact('age_clean', 'age', label='age_exact') features = compare_cl.compute(candidate_links, df_raw, df_erp) # 3. 分类（用简单规则，非ML） # 规则：手机号精确匹配，或（姓名相似度>0.85 且 年龄相同） matches = features[ (features['phone_exact'] == 1) | ((features['name_sim'] == 1) & (features['age_exact'] == 1)) ].index # 4. 合并逻辑：取raw表的customer_id为主键，合并erp表的字段 df_merged = df_raw.copy() for idx_left, idx_right in matches: # 用erp表的更完整地址覆盖raw表 if pd.isna(df_merged.loc[idx_left, 'address_clean']) and pd.notna(df_erp.loc[idx_right, 'full_address']): df_merged.loc[idx_left, 'address_clean'] = df_erp.loc[idx_right, 'full_address'] # 合并来源标记 df_merged.loc[idx_left, 'source_systems'] = f"{df_merged.loc[idx_left, 'source_systems']},ERP" # 5. 输出主数据表 df_merged.to_parquet("customer_master_v202310.parquet", index=False) print(f"Merged {len(matches)} records. Final master table has {len(df_merged)} rows.")

避坑技巧：

• indexer.block必须用业务有意义的字段。用hash(customer_id)会完全失效，因为两系统ID不同。
• compare_cl.exact('age_clean', 'age')：exact比string快10倍，且年龄是数值，无需模糊匹配。
• 合并时“覆盖逻辑”必须业务确认。我们约定：ERP地址更权威，CRM手机号更及时，所以优先用ERP地址，CRM手机号。

5.4 第四步：治理——用Great Expectations把质量关死

pip install great_expectations great_expectations init # 按向导初始化

创建great_expectations/yaml_config.yml：

datasources: pandas_datasource: class_name: PandasDatasource batch_kwargs_generators: subdir_reader: class_name: SubdirReaderBatchKwargsGenerator base_directory: ./data/ validation_operators: action_list_operator: class_name: ActionListValidationOperator action_list: - name: store_validation_result action: class_name: StoreValidationResultAction - name: store_evaluation_params action: class_name: StoreEvaluationParametersAction - name: update_data_docs action: class_name: UpdateDataDocsAction # 定义期望 expectations: - expectation_suite_name: customer_master_suite data_asset_type: PandasDataset expectations: - expectation_type: expect_table_row_count_to_be_between kwargs: min_value: 8500 max_value: 9000 - expectation_type: expect_column_values_to_not_be_null kwargs: column: customer_id - expectation_type: expect_column_values_to_match_regex kwargs: column: phone_clean regex: '^\d{11}$' - expectation_type: expect_column_values_to_be_between kwargs: column: age_clean min_value: 0 max_value: 120 - expectation_type: expect_column_proportion_of_unique_values_to_be_between kwargs: column: customer_id