以下描述从主数据管理（MDM）系统到数据仓库维度表 dim_cust 的端到端数据血缘关系，覆盖数据摄取、转换、存储与供数流程，并给出关键转换规则与实现示例。
范围与目标
- 目标对象：dim_cust（客户维度表，支持事实表进行客户维度关联与分析）
- 设计原则：以 MDM 的“黄金客户”主记录为准，采用 SCD Type 2 管理可历史化属性，保留列级血缘与作业级血缘元数据
- 关键键：
- 业务键：mdm_customer_id（来自 MDM 的统一客户主键）
- 代理键：cust_sk（数据仓库内生成的 surrogate key）
- 时间列：effective_start_dt, effective_end_dt, current_flag
血缘分层与流程
- 源系统（MDM）
- 主要实体表（示例命名，与具体实现可有差异）：
- MDM_CUSTOMER：客户主档（customer_id, name, type_code, status, effective_start_ts, effective_end_ts, source_system, record_version 等）
- MDM_ADDRESS：地址（customer_id, address_type, country_code, region, city, postal_code, is_primary 等）
- MDM_CONTACT：联系方式（customer_id, email, phone, is_primary, contact_status 等）
- MDM_XREF（可选）：跨源映射（customer_id 与各源系统的原始键映射）
- 输出机制：建议以 CDC（变更数据捕获）为主，包含插入、更新、删除（软/硬）事件；如无法 CDC，使用每日快照文件并计算增量
- 着陆层（Raw/Bronze）
- 摄取：通过 Kafka Connect + Debezium（或数据库原生日志订阅/变更表），将 MDM_* 的变更事件流入消息队列，再落地到对象存储/数据湖（Parquet/Avro）
- 表示：
- raw_mdm_customer：与源模式一致，包含操作类型（op_type）、事务时间（op_ts）、源事务键（src_tx_id）
- raw_mdm_address、raw_mdm_contact：同上
- 保留策略：原样保留、不可变存档，按 ingestion_date 分区，作为审计与重放依据
- 标准化层（Staging/Silver）
- 处理目标：模式统一、类型规范化、码值映射、主记录去重、主从实体汇总
- 产出：
- stg_customer：每个 mdm_customer_id 保留最新有效版本（基于 effective_* 与 record_version），规范化字段（如统一国家码、名称清洗）
- stg_customer_contacts：主邮箱/主电话解析（优先 is_primary=true，且 contact_status=active）
- stg_customer_address：主地址解析（优先 address_type='PRIMARY' 或 is_primary=true）
- 变更识别：为支持后续 SCD2，计算变更哈希（hash_diff）仅覆盖需历史化的属性集
- 统一与整备层（Conformed/Model）
- 合并主档与主地址/主联系方式，形成统一客户核心视图 cust_core
- 规则：
- 单值选择：若多个主标识冲突，按优先级（is_primary、最近生效时间、来源可信度）选一
- 状态过滤：仅保留 MDM 标记为 active 且当前生效的客户记录；停用或过期记录保留用于关闭 SCD2
- 维度构建层（Warehouse/Gold）
- 目标表：dim_cust
- SCD 策略：
- Type 2（历史化）：客户名称、类型、国家码、主地址、主联系方式等业务分析敏感属性
- Type 1（覆盖）：数据质量标签、技术元数据（如标准化标记），必要时进行覆盖更新
- 合并逻辑：
- 新客户：插入新维度行（生成 cust_sk，设置 current_flag=1）
- 属性变更：关闭旧行（设置 effective_end_dt、current_flag=0），插入新行（current_flag=1）
- 停用/删除：根据 MDM 的 status 或硬删除事件，关闭当前行并标注 end_reason
- 存储：列式数据仓库（Snowflake/BigQuery/Redshift/Databricks Delta），分区按有效期或加载日期
- 下游消费（Marts/事实表/服务）
- 事实表使用 cust_sk 进行维度外键关联
- 数据服务与报表（BI/客服系统）通过维度获取客户主档与历史快照
列级血缘与转换规则（示例）
- dim_cust.cust_sk：仓库内生成（序列/IDENTITY/UUID），无上游列
- dim_cust.mdm_customer_id：来源 MDM_CUSTOMER.customer_id（stg_customer.customer_id）
- dim_cust.customer_name：来源 MDM_CUSTOMER.name，经名称清洗（去除多余空格/大小写规范）
- dim_cust.customer_type：来源 MDM_CUSTOMER.type_code，经码值映射（例如 “P”->“个人”，“B”->“企业”）
- dim_cust.primary_email：来源 stg_customer_contacts.email，规则 is_primary=true 且 contact_status=active
- dim_cust.primary_phone：来源 stg_customer_contacts.phone，规则同上
- dim_cust.country_code：来源 stg_customer_address.country_code，经 ISO 3166-1 alpha-2 校正
- dim_cust.address_line_1/…：来源 stg_customer_address.*，地址标准化（缩写展开、邮编校验）
- dim_cust.is_active：来源 MDM_CUSTOMER.status in ('ACTIVE') 且未过期（effective_end_ts is null 或 > 业务时间）
- dim_cust.effective_start_dt/effective_end_dt/current_flag：由 SCD2 合并流程生成
- dim_cust.src_system：来源 MDM_CUSTOMER.source_system
- dim_cust.src_row_id：来源 MDM_CUSTOMER.source_row_id（如存在），用于溯源到原始源系统
- dim_cust.hash_diff：对需历史化的列按稳定序列计算哈希（如 SHA256(name, type, country_code, address_hash, email, phone)）
- dim_cust.load_dt/update_dt：ETL 作业时间戳
- dim_cust.end_reason（可选）：'INACTIVE'/'DELETED' 等，从 MDM 事件或状态派生
SCD2 合并逻辑示例（SQL 通用伪代码）
注意：不同仓库的 MERGE 语法略有差异，以下为通用思路。
-- 变更检测视图：仅保留当前有效的 MDM 记录，并计算 hash_diff
WITH src AS (
SELECT
c.customer_id AS mdm_customer_id,
norm_name(c.name) AS customer_name,
map_type(c.type_code) AS customer_type,
a.country_code AS country_code,
a.address_line_1,
a.address_line_2,
ct.email AS primary_email,
ct.phone AS primary_phone,
c.status AS mdm_status,
c.effective_start_ts AS src_effective_start_ts,
c.effective_end_ts AS src_effective_end_ts,
c.source_system AS src_system,
c.source_row_id AS src_row_id,
sha2(concat_ws('||',
norm_name(c.name),
map_type(c.type_code),
a.country_code,
coalesce(a.address_line_1,''),
coalesce(a.address_line_2,''),
coalesce(ct.email,''),
coalesce(ct.phone,'')
), 256) AS hash_diff,
current_timestamp() AS load_dt
FROM stg_customer c
LEFT JOIN stg_customer_address a
ON a.customer_id = c.customer_id AND a.is_primary = true
LEFT JOIN stg_customer_contacts ct
ON ct.customer_id = c.customer_id AND ct.is_primary = true AND ct.contact_status = 'ACTIVE'
WHERE (c.effective_end_ts IS NULL OR c.effective_end_ts > current_date())
)
-- 关闭发生变化的旧版本
UPDATE dim_cust AS d
SET effective_end_dt = current_date(),
current_flag = 0,
update_dt = current_timestamp(),
end_reason = CASE WHEN src.mdm_status <> 'ACTIVE' THEN 'INACTIVE' ELSE 'CHANGED' END
FROM src
WHERE d.mdm_customer_id = src.mdm_customer_id
AND d.current_flag = 1
AND d.hash_diff <> src.hash_diff;
-- 插入新版本（新客户或属性变更后版本）
INSERT INTO dim_cust (
cust_sk,
mdm_customer_id,
customer_name,
customer_type,
country_code,
address_line_1,
address_line_2,
primary_email,
primary_phone,
is_active,
effective_start_dt,
effective_end_dt,
current_flag,
src_system,
src_row_id,
hash_diff,
load_dt,
update_dt
)
SELECT
gen_surrogate_key(), -- 仓库侧生成
s.mdm_customer_id,
s.customer_name,
s.customer_type,
s.country_code,
s.address_line_1,
s.address_line_2,
s.primary_email,
s.primary_phone,
CASE WHEN s.mdm_status = 'ACTIVE' THEN 1 ELSE 0 END,
current_date(),
NULL,
1,
s.src_system,
s.src_row_id,
s.hash_diff,
s.load_dt,
s.load_dt
FROM src s
LEFT JOIN dim_cust d
ON d.mdm_customer_id = s.mdm_customer_id AND d.current_flag = 1
WHERE d.mdm_customer_id IS NULL -- 新客户
OR d.hash_diff <> s.hash_diff; -- 发生属性变更
删除/停用处理
- 软删除或停用（MDM.status='INACTIVE' 或 src_effective_end_ts 到期）：关闭当前行并设置 end_reason='INACTIVE'
- 硬删除事件：关闭当前行并设置 end_reason='DELETED'；不物理删除维度历史行
数据质量与治理要点
- 唯一性：mdm_customer_id 全局唯一；dim_cust 中 current_flag=1 的 mdm_customer_id 唯一
- 完整性：必填列（mdm_customer_id, customer_name, customer_type, country_code）非空
- 码值校验：type_code、country_code 必须在维护的参照表内
- PII 管理：email/phone 标记敏感；落地层加密或脱敏；使用列级访问控制
- 审计列：load_dt, update_dt, src_system, src_row_id, op_ts（可于原始层保留）
- 变更可追溯：保留 raw 层原始事件与 stg 层标准化快照；维度层保存历史行
元数据与血缘登记（OpenLineage 作业级示例，简化版）
{
"eventType": "COMPLETE",
"job": {
"namespace": "dwh.customer",
"name": "build_dim_cust_scd2"
},
"run": { "runId": "a1b2c3d4-e5f6-7890-1234-abcdef987654" },
"inputs": [
{ "namespace": "lake.raw", "name": "raw_mdm_customer" },
{ "namespace": "lake.raw", "name": "raw_mdm_address" },
{ "namespace": "lake.raw", "name": "raw_mdm_contact" }
],
"outputs": [
{ "namespace": "warehouse.gold", "name": "dim_cust" }
],
"producer": "airflow://dags/build_dim_cust",
"schemaVersion": "1-0-0"
}
调度与依赖
- DAG 顺序：
- ingest_mdm_cdc → raw
- standardize_mdm_customer → stg_customer/stg_address/stg_contacts
- conform_customer_core → cust_core
- scd2_dim_cust_merge → dim_cust
- 频率：与 MDM 事件频率一致（近实时或批次），维度合并可按小时/日运行
- 失败重试：以 raw 层为重放来源，幂等合并（基于 mdm_customer_id 与 hash_diff）
总结
- 血缘从 MDM（MDM_CUSTOMER/ADDRESS/CONTACT）经 Raw→Staging→Conformed→Warehouse（dim_cust）逐层推进
- 每个列在 dim_cust 中可回溯到对应的 MDM 源列与标准化/映射规则
- 作业级血缘通过编排工具与 OpenLineage/Atlas 等登记，支持端到端追踪与审计
- 采用 SCD2 管理客户属性历史，保证事实表与报表的时点一致性与可解释性
