2026 技术观察：企业主数据治理进入智能化阶段，实体去重、标准编码和可信合并成为数据中台新基础

原创

用户12583401

发布于 2026-07-11 12:54:09

260

概述

2026 年，企业数据治理正在从“建设更多数据表”走向“统一核心业务实体”。

在很多企业中，同一个客户可能同时存在于 CRM、订单系统、客服系统、会员平台和财务系统中。

由于录入方式、字段格式和系统编码不同，同一个客户可能出现多个名称。

例如，“上海云启科技有限公司”“上海云启科技”和“云启科技上海公司”，可能实际代表同一主体。

如果这些记录没有统一，企业就难以准确统计客户数量、交易金额、服务记录和信用风险。

因此，企业主数据治理开始进入智能化阶段。

系统需要自动完成格式标准化、相似实体识别、重复记录合并和统一编码，为数据分析、业务系统和 AI 应用提供可信基础数据。

一、为什么企业需要主数据治理？

主数据是多个系统共同使用的核心业务对象。

常见主数据包括客户、供应商、商品、员工、组织、物料和门店。

主数据治理系统可以帮助企业回答几个问题：

哪些记录属于同一个客户；
哪个系统中的字段更加可信；
如何生成统一客户编码；
哪些字段存在冲突；
合并后如何保留来源记录；
如何持续监测新增重复数据。

下面用 Python 写一个简化版客户主数据治理系统。

二、基础数据：定义多系统客户记录

第一步是准备来自不同系统的客户数据。

import json
import re
from difflib import SequenceMatcher
from datetime import datetime
from collections import defaultdict


RAW_CUSTOMERS = [
    {
        "source": "crm",
        "source_id": "CRM001",
        "name": "上海云启科技有限公司",
        "phone": "021-88889999",
        "email": "service@yunqi.example",
        "city": "上海市"
    },
    {
        "source": "order",
        "source_id": "ORD109",
        "name": "上海云启科技",
        "phone": "02188889999",
        "email": "service@yunqi.example",
        "city": "上海"
    },
    {
        "source": "support",
        "source_id": "SUP026",
        "name": "云启科技（上海）",
        "phone": "021-8888-9999",
        "email": "",
        "city": "上海"
    },
    {
        "source": "crm",
        "source_id": "CRM002",
        "name": "杭州星河智能技术有限公司",
        "phone": "0571-66668888",
        "email": "contact@xinghe.example",
        "city": "杭州"
    }
]

这些记录存在名称、电话和城市格式差异。

主数据治理需要先完成标准化，再进行实体匹配。

三、字段标准化处理

第二步是统一公司名称、电话和城市格式。

COMPANY_SUFFIXES = [
    "有限责任公司",
    "股份有限公司",
    "有限公司",
    "公司"
]


def normalize_company_name(name):
    value = name.strip().lower()

    value = re.sub(
        r"[（）()\s]",
        "",
        value
    )

    for suffix in COMPANY_SUFFIXES:
        value = value.replace(
            suffix,
            ""
        )

    return value


def normalize_phone(phone):
    return re.sub(
        r"\D",
        "",
        phone or ""
    )


def normalize_city(city):
    return (
        city.replace("市", "")
        .replace("省", "")
        .strip()
    )


def normalize_customer(customer):
    result = customer.copy()

    result["normalized_name"] = normalize_company_name(
        customer["name"]
    )

    result["normalized_phone"] = normalize_phone(
        customer["phone"]
    )

    result["normalized_city"] = normalize_city(
        customer["city"]
    )

    result["normalized_email"] = (
        customer["email"]
        .strip()
        .lower()
    )

    return result

标准化可以解决大量格式差异。

电话中的横线、公司名称中的括号和城市后缀，都不应该影响实体判断。

四、实体相似度计算

第三步是综合名称、电话、邮箱和城市计算相似度。

def calculate_customer_similarity(
    customer_a,
    customer_b
):
    name_score = SequenceMatcher(
        None,
        customer_a["normalized_name"],
        customer_b["normalized_name"]
    ).ratio()

    score = name_score * 50
    reasons = [
        f"名称相似度为 {round(name_score, 2)}。"
    ]

    if (
        customer_a["normalized_phone"]
        and customer_a["normalized_phone"]
        == customer_b["normalized_phone"]
    ):
        score += 25
        reasons.append("联系电话一致。")

    if (
        customer_a["normalized_email"]
        and customer_a["normalized_email"]
        == customer_b["normalized_email"]
    ):
        score += 20
        reasons.append("电子邮箱一致。")

    if (
        customer_a["normalized_city"]
        == customer_b["normalized_city"]
    ):
        score += 5
        reasons.append("所在城市一致。")

    return {
        "score": round(min(score, 100), 2),
        "reasons": reasons
    }

实体匹配不能只依赖名称。

名称可能相似但并非同一企业，电话或邮箱一致通常可以提供更强证据。

五、重复实体分组

第四步是把高相似度记录归入同一个实体组。

def detect_duplicate_customer_groups(
    customers,
    threshold=75
):
    groups = []
    used_ids = set()

    for customer in customers:
        source_id = customer["source_id"]

        if source_id in used_ids:
            continue

        current_group = [customer]
        used_ids.add(source_id)

        for candidate in customers:
            candidate_id = candidate["source_id"]

            if candidate_id in used_ids:
                continue

            similarity = calculate_customer_similarity(
                customer,
                candidate
            )

            if similarity["score"] >= threshold:
                current_group.append(candidate)
                used_ids.add(candidate_id)

        groups.append(current_group)

    return groups

实体分组是主数据合并的前一步。

相似度不够高的记录不应该自动合并，而应进入人工复核队列。

六、主记录字段合并

第五步是从多条来源记录中生成可信主记录。

SOURCE_PRIORITY = {
    "crm": 3,
    "order": 2,
    "support": 1
}


def choose_best_value(records, field):
    candidates = [
        record
        for record in records
        if record.get(field)
    ]

    if not candidates:
        return ""

    candidates.sort(
        key=lambda item: SOURCE_PRIORITY.get(
            item["source"],
            0
        ),
        reverse=True
    )

    return candidates[0][field]


def merge_master_customer(
    group,
    master_index
):
    master_id = f"CUSTOMER_{master_index:05d}"

    return {
        "master_id": master_id,
        "name": choose_best_value(group, "name"),
        "phone": choose_best_value(group, "phone"),
        "email": choose_best_value(group, "email"),
        "city": choose_best_value(group, "city"),
        "source_records": [
            {
                "source": record["source"],
                "source_id": record["source_id"]
            }
            for record in group
        ],
        "source_count": len(group)
    }

字段合并需要定义可信来源优先级。

例如客户名称以 CRM 为准，交易地址可能以最新订单为准，具体规则需要结合企业业务设计。

七、字段冲突检查

第六步是检查合并组中是否存在重要字段冲突。

def detect_master_data_conflicts(group):
    conflicts = []

    fields = [
        "normalized_phone",
        "normalized_email",
        "normalized_city"
    ]

    for field in fields:
        values = {
            record[field]
            for record in group
            if record.get(field)
        }

        if len(values) > 1:
            conflicts.append({
                "field": field,
                "values": sorted(values),
                "message": "同一实体组内存在字段冲突。"
            })

    return conflicts

名称相似并不意味着所有字段都一致。

如果同一组中存在多个不同电话或邮箱，就需要进一步判断是否发生了数据录入错误。

八、主数据质量评分

第七步是对生成后的主记录进行质量评估。

def evaluate_master_data_quality(
    master_record,
    conflicts
):
    score = 100
    issues = []

    required_fields = [
        "name",
        "phone",
        "city"
    ]

    for field in required_fields:
        if not master_record.get(field):
            score -= 15
            issues.append(
                f"关键字段 {field} 缺失。"
            )

    if conflicts:
        score -= len(conflicts) * 10
        issues.append("主实体存在来源字段冲突。")

    if master_record["source_count"] >= 2:
        score += 5

    score = max(
        min(score, 100),
        0
    )

    if score >= 90:
        level = "trusted"
    elif score >= 75:
        level = "usable"
    elif score >= 60:
        level = "attention"
    else:
        level = "poor"

    return {
        "master_id": master_record["master_id"],
        "quality_score": score,
        "quality_level": level,
        "issues": issues
    }

主数据质量评分可以帮助企业区分可信记录和待治理记录。

低质量主记录不应直接进入关键分析和自动化流程。

九、运行完整主数据治理流程

最后生成统一客户主数据报告。

def run_master_data_governance():
    normalized_customers = [
        normalize_customer(customer)
        for customer in RAW_CUSTOMERS
    ]

    groups = detect_duplicate_customer_groups(
        normalized_customers
    )

    master_records = []
    conflict_results = []
    quality_results = []

    for index, group in enumerate(
        groups,
        start=1
    ):
        master_record = merge_master_customer(
            group,
            index
        )

        conflicts = detect_master_data_conflicts(
            group
        )

        quality = evaluate_master_data_quality(
            master_record,
            conflicts
        )

        master_records.append(master_record)

        conflict_results.append({
            "master_id": master_record["master_id"],
            "conflicts": conflicts
        })

        quality_results.append(quality)

    report = {
        "report_name": "企业客户主数据治理报告",
        "raw_record_count": len(RAW_CUSTOMERS),
        "master_record_count": len(master_records),
        "duplicate_reduction_count": (
            len(RAW_CUSTOMERS)
            - len(master_records)
        ),
        "master_records": master_records,
        "conflict_results": conflict_results,
        "quality_results": 30655.t.kuaisou.com 
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_master_data_governance()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))

十、趋势判断

从这套流程可以看到，企业主数据治理正在从人工整理走向智能实体融合。

未来，数据中台不会只汇总不同系统的数据，还需要判断哪些记录代表同一个业务对象，并为其生成统一身份编码。

主数据越可靠，客户分析、供应链管理、财务核算和 AI 应用的结果就越可信。

谁能把字段标准化、实体匹配、可信合并和质量监控结合起来，谁就更容易建立统一、稳定的企业数据基础。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

数据库

企业

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

数据库

企业

登录后参与评论

0 条评论

热度