以下为表 fact_orders 的列 order_status 的“枚举值校验”可执行规则与检查步骤（含示例代码）。目标是确保该字段仅取自统一状态集，规范大小写与空白，避免未知占位与同义异写，支撑履约、退款与收入确认的一致性。
一、校验规则（可操作定义）
- 允许值全集（去首尾空格后、统一大写判定）：
{PENDING, PAID, SHIPPED, DELIVERED, CANCELLED, REFUNDED, FAILED_PAYMENT}
- 规则细则：
- 预处理：对原值执行 trim 和 upper，最终标准化为大写存储。
- 非法值：空值、空串、仅空白、NULL 字符串、未知占位（N/A、NA、NONE、-、UNKNOWN 等）均判定违规。
- 枚举校验：预处理后值不在允许集合即为违规。
- 维表约束：建立维表 dim_order_status 存放枚举与释义；入仓时进行字典对齐并仅允许维表内值写入事实表。
- 建议规范化管控：全链路统一使用标准化函数 normalize_status() 输出最终持久化值。
二、检查要点与统计输出
- A) 预处理：trim + upper，剔除无意义占位。
- B) 枚举校验：not in 列表即违规。
- C) 统计输出：
- 各枚举值分布（计数与占比）。
- 未知值 TopN（原始值及其计数）。
- 违规总数与占比（违规计数/总计数）。
三、异常处理与自动修复策略
- 违规落库：将违规记录入库至 bad_records.fact_orders_order_status，字段建议包含：
order_id, raw_order_status, normalized_status, violation_reason, recommended_mapping, source_system, etl_batch_id, processed_at。
- 推荐映射：若能识别常见拼写或变体（如 Paid、paid、paid 、shiped、shipped ），给出 recommended_mapping。
- 自动修复门槛：
- 低风险（违规占比 ≤ 0.05% 且为明显小写/空格/常见拼写）：在入仓修正为标准值，并记录修正日志。
- 高风险（> 0.05% 或无法可靠映射）：维持为未知分组，不写入事实表目标列；需业务确认后回填。
四、实施建议
- 在数据接入层引入“字典映射组件”（基于维表与同义词表），在落仓前统一标准化。
- 在报表层对未知状态设置兜底分组“UNKNOWN_STATUS”，并实时展示修复指引（TopN 未知值与建议映射）。
- 与业务状态机和事件流对齐，设置状态跃迁校验（例如禁止从 DELIVERED 回退至 PAID）。
五、示例代码
- 维表与同义词表（ANSI 风格 SQL）
-- 维表：标准字典
CREATE TABLE IF NOT EXISTS dim_order_status (
status_code VARCHAR(32) PRIMARY KEY,
status_desc VARCHAR(256)
);
MERGE INTO dim_order_status d
USING (
SELECT 'PENDING' AS c, '待支付/待处理' AS s UNION ALL
SELECT 'PAID', '已支付' UNION ALL
SELECT 'SHIPPED', '已发货' UNION ALL
SELECT 'DELIVERED', '已送达/签收' UNION ALL
SELECT 'CANCELLED', '已取消' UNION ALL
SELECT 'REFUNDED', '已退款' UNION ALL
SELECT 'FAILED_PAYMENT', '支付失败'
) src
ON d.status_code = src.c
WHEN NOT MATCHED THEN INSERT (status_code, status_desc) VALUES (src.c, src.s);
-- 可选：同义词/常见拼写映射表（优先精确/正则，避免过度泛化）
CREATE TABLE IF NOT EXISTS ref_order_status_synonyms (
pattern VARCHAR(64), -- 匹配模式（存放大写+去空格的形态）
mapped_to VARCHAR(32), -- 目标标准值
match_type VARCHAR(16) -- EXACT 或 REGEX
);
INSERT INTO ref_order_status_synonyms (pattern, mapped_to, match_type) VALUES
('PAID', 'PAID', 'EXACT'),
('PENDING', 'PENDING', 'EXACT'),
('SHIPPED', 'SHIPPED', 'EXACT'),
('DELIVERED', 'DELIVERED', 'EXACT'),
('CANCELLED', 'CANCELLED', 'EXACT'),
('REFUNDED', 'REFUNDED', 'EXACT'),
('FAILEDPAYMENT', 'FAILED_PAYMENT', 'EXACT'),
-- 常见变体（存放去空格+大写后形态）
('PAID', 'PAID', 'EXACT'),
('PAYED', 'PAID', 'EXACT'),
('SHIPED', 'SHIPPED', 'EXACT'),
('SHIPPEDD', 'SHIPPED', 'EXACT'),
('CANCELED', 'CANCELLED', 'EXACT'),
('FAILED_PAYMENT', 'FAILED_PAYMENT', 'EXACT');
- 预处理与枚举校验（ANSI 风格 SQL）
-- 参数：:batch_id 可替换为实际批次号
WITH pre AS (
SELECT
o.,
TRIM(o.order_status) AS trimmed_status,
UPPER(TRIM(o.order_status)) AS upper_status,
REPLACE(UPPER(TRIM(o.order_status)),' ','') AS compact_status,
CASE
WHEN o.order_status IS NULL THEN 'NULL'
WHEN TRIM(o.order_status) = '' THEN 'EMPTY'
WHEN UPPER(TRIM(o.order_status)) IN ('NULL','N/A','NA','NONE','-','UNKNOWN') THEN 'PLACEHOLDER'
ELSE NULL
END AS nullish_reason
FROM fact_orders o
),
norm AS (
-- 基于同义词进行一次映射；未命中则保留 upper_status
SELECT
p.,
COALESCE(s.mapped_to, p.upper_status) AS normalized_status
FROM pre p
LEFT JOIN ref_order_status_synonyms s
ON p.compact_status = s.pattern AND s.match_type = 'EXACT'
),
chk AS (
SELECT
n.,
CASE
WHEN nullish_reason IS NOT NULL THEN 'NULL_OR_PLACEHOLDER'
WHEN d.status_code IS NULL THEN 'NOT_IN_ENUM'
ELSE NULL
END AS violation_reason
FROM norm n
LEFT JOIN dim_order_status d
ON n.normalized_status = d.status_code
)
-- 统计输出
SELECT
SUM(CASE WHEN violation_reason IS NULL THEN 1 ELSE 0 END) AS valid_cnt,
SUM(CASE WHEN violation_reason IS NOT NULL THEN 1 ELSE 0 END) AS invalid_cnt,
COUNT() AS total_cnt,
ROUND(100.0 * SUM(CASE WHEN violation_reason IS NOT NULL THEN 1 ELSE 0 END) / NULLIF(COUNT(*),0), 4) AS invalid_pct
FROM chk;
-- 有效值分布
SELECT normalized_status AS status_code, COUNT(*) AS cnt
FROM chk
WHERE violation_reason IS NULL
GROUP BY normalized_status
ORDER BY cnt DESC;
-- 未知/违规 TopN
SELECT
COALESCE(upper_status, '(NULL)') AS raw_upper_status,
COUNT(*) AS cnt,
MIN(violation_reason) AS reason
FROM chk
WHERE violation_reason IS NOT NULL
GROUP BY COALESCE(upper_status, '(NULL)')
ORDER BY cnt DESC
FETCH FIRST 50 ROWS WITH TIES;
- 违规落库与推荐映射（ANSI 风格 SQL）
-- 建议先建宽表用于审计
CREATE TABLE IF NOT EXISTS bad_records.fact_orders_order_status (
order_id VARCHAR(128),
raw_order_status VARCHAR(256),
normalized_status VARCHAR(64),
violation_reason VARCHAR(64),
recommended_mapping VARCHAR(64),
source_system VARCHAR(128),
etl_batch_id VARCHAR(64),
processed_at TIMESTAMP
);
-- 插入违规记录（recommended_mapping：若 enum 中不存在且同义词表命中则给出；否则 NULL）
INSERT INTO bad_records.fact_orders_order_status
SELECT
c.order_id,
c.order_status AS raw_order_status,
c.normalized_status,
c.violation_reason,
CASE
WHEN c.violation_reason = 'NOT_IN_ENUM' THEN rs2.mapped_to
ELSE NULL
END AS recommended_mapping,
c.source_system,
:batch_id AS etl_batch_id,
CURRENT_TIMESTAMP AS processed_at
FROM chk c
LEFT JOIN ref_order_status_synonyms rs2
ON c.compact_status = rs2.pattern AND rs2.match_type = 'EXACT'
WHERE c.violation_reason IS NOT NULL;
- 低风险自动修复（只更正大小写/空格/已知常见拼写，且当整体违规占比 ≤ 0.05%）
-- 计算违规占比
WITH stats AS (
SELECT
SUM(CASE WHEN violation_reason IS NOT NULL THEN 1 ELSE 0 END) AS invalid_cnt,
COUNT(*) AS total_cnt
FROM chk
)
-- 示例：仅当门槛满足时，对可安全映射的数据进行就地更新（请根据目标仓库是否允许 UPDATE 调整实现）
UPDATE fact_orders t
SET order_status = c.normalized_status
FROM chk c, stats s
WHERE t.order_id = c.order_id
AND c.violation_reason IN ('NOT_IN_ENUM') -- 仅处理可映射的轻微问题
AND EXISTS (SELECT 1 FROM dim_order_status d WHERE d.status_code = c.normalized_status)
AND (1.0 * s.invalid_cnt / NULLIF(s.total_cnt, 0)) <= 0.0005;
-- 无法安全修复的记录保留在 bad_records，待业务确认后回填
- 数据入仓的统一标准化函数（PySpark，批/流皆可）
from pyspark.sql import functions as F, types as T
ALLOWED = {"PENDING","PAID","SHIPPED","DELIVERED","CANCELLED","REFUNDED","FAILED_PAYMENT"}
PLACEHOLDERS = {"NULL","N/A","NA","NONE","-","UNKNOWN"}
COMMON_MAP = {
"PAYED": "PAID",
"SHIPED": "SHIPPED",
"SHIPPEDD": "SHIPPED",
"CANCELED": "CANCELLED",
"FAILED PAYMENT": "FAILED_PAYMENT",
"FAILEDPAYMENT": "FAILED_PAYMENT",
}
@F.udf("string")
def normalize_status(s):
if s is None:
return None
t = s.strip()
if t == "":
return None
up = t.upper()
if up in PLACEHOLDERS:
return None
compact = up.replace(" ", "")
if compact in ALLOWED:
return compact
if compact in COMMON_MAP:
return COMMON_MAP[compact]
# 保守策略：不做模糊修复，避免误判；如需，可增加 levenshtein <= 1 的映射白名单
return None
def with_status_validation(df):
df1 = df.withColumn("normalized_status", normalize_status(F.col("order_status")))
df_valid = df1.filter(F.col("normalized_status").isin(list(ALLOWED)))
df_invalid = df1.filter(~F.col("normalized_status").isin(list(ALLOWED)))
return df_valid, df_invalid
使用示例
df_orders: 输入 DataFrame，包含 order_id, order_status, source_system 等
df_valid, df_invalid = with_status_validation(df_orders)
统计
df_stats = df1.groupBy("normalized_status").count()
落库违规
df_invalid
.withColumn("violation_reason",
F.when(F.col("normalized_status").isNull(), F.lit("NULL_OR_PLACEHOLDER"))
.otherwise(F.lit("NOT_IN_ENUM")))
.withColumn("etl_batch_id", F.lit(batch_id))
.withColumn("processed_at", F.current_timestamp())
.write.mode("append").insertInto("bad_records.fact_orders_order_status")
将 df_valid 写入事实表（注意仅写 normalized_status）
df_valid.select(
"order_id",
F.col("normalized_status").alias("order_status"),
"source_system",
# 其他业务列...
).write.mode("append").insertInto("fact_orders")
- dbt 通用测试（可复用的 generic test）
-- models/schema.yml
version: 2
models:
- name: fact_orders
columns:
- name: order_status
tests:
- not_null
- accepted_values:
values: ['PENDING','PAID','SHIPPED','DELIVERED','CANCELLED','REFUNDED','FAILED_PAYMENT']
-- 若需标准化，可在 staging 层构建视图：upper(trim(order_status)) as order_status，然后在该视图上做 accepted_values
六、部署与运行顺序建议
- 在接入层或 Staging 层执行 normalize（trim + upper + 同义词映射），并产出 normalized_status。
- 用维表 dim_order_status 做枚举校验与维度对齐；禁止非维表值进入事实表最终列。
- 生成统计输出（分布、未知 TopN、违规率）。
- 将违规数据落库 bad_records.fact_orders_order_status，并带上推荐映射。
- 若违规率 ≤ 0.05%，仅执行安全自动修复（小写、空格、常见拼写）；否则输出修复清单，待业务确认后回填。
- 在报表层对未知状态设置兜底分组并显示修复指引。
- 持续对齐业务状态机与事件事件表（可另设状态跃迁约束检查）。
