PySpark data validation for fact_order_item.total_amount
Rules implemented:
- 非空: forbid NULL and empty string; canceled orders must use 0 not NULL.
- 数据类型匹配: normalize to DECIMAL(18,2), two decimals via HALF_UP rounding; forbid scientific notation and currency symbols; clean thousands separator and non-digit chars; numeric-ize before write.
- 范围校验: 0 <= total_amount <= 1_000_000; negatives not allowed; out-of-range marked for business review.
- 依赖关系校验: total_amount ≈ quantityunit_price within 0.01; if discount_amount/tax_amount present, total_amount = quantityunit_price - discount_amount + tax_amount. Sample-based verification supported for batch runs.
- 一致性: same natural key across partitions/batches must be consistent in amount.
- 异常处理: output detailed error records for non-numeric/overflow/consistency issues.
from typing import Dict, List, Optional
from pyspark.sql import SparkSession, DataFrame, functions as F, types as T
spark = SparkSession.builder.getOrCreate()
AMOUNT_UPPER_LIMIT = 1_000_000.0
AMOUNT_TOLERANCE = 0.01
DECIMAL_TYPE = T.DecimalType(18, 2)
def has_cols(df: DataFrame, cols: List[str]) -> bool:
return all(c in df.columns for c in cols)
def first_existing(df: DataFrame, candidates: List[str]) -> Optional[str]:
for c in candidates:
if c in df.columns:
return c
return None
def normalize_total_amount(df_raw: DataFrame, total_col: str = "total_amount") -> DataFrame:
"""
1) Strip thousand separators and currency symbols.
2) Forbid scientific notation; mark as invalid if contains e/E.
3) Enforce numeric pattern ^-?\d+(\.\d+)?$
4) Cast to DOUBLE, round HALF_UP to 2 decimals, cast to DECIMAL(18,2).
Returns df with:
- total_amount_clean_str
- total_amount_num_dec (DECIMAL(18,2)) normalized
- flags: invalid_number, sci_notation_found
"""
# capture raw string for diagnostics
df = df_raw.withColumn("total_amount_raw_str", F.col(total_col).cast("string"))
# detect scientific notation beforehand
df = df.withColumn(
"sci_notation_found",
F.when(F.col("total_amount_raw_str").rlike("(?i)[+-]?[0-9]*\\.?[0-9]+[eE][+-]?[0-9]+"), F.lit(True)).otherwise(F.lit(False)),
)
# remove common currency symbols and thousand separators, spaces
# keep digits, dot, minus
df = df.withColumn(
"total_amount_clean_str",
F.regexp_replace(F.col("total_amount_raw_str"), r"[,\s¥￥$€£]", "")
).withColumn(
"total_amount_clean_str",
F.regexp_replace(F.col("total_amount_clean_str"), r"[^0-9\.\-]", "")
)
# validate simple decimal pattern (no scientific notation)
numeric_pattern = r"^-?\d+(\.\d+)?$"
df = df.withColumn(
"numeric_like",
F.when(F.col("total_amount_clean_str").rlike(numeric_pattern), F.lit(True)).otherwise(F.lit(False))
)
# mark invalid numbers (include sci notation or bad pattern)
df = df.withColumn(
"invalid_number",
(~F.col("numeric_like")) | F.col("sci_notation_found")
)
# cast to double and round HALF_UP using round(), then cast to DECIMAL(18,2)
# invalid_number will yield null after cast; keep for error reporting
df = df.withColumn(
"total_amount_num_dbl",
F.when(~F.col("invalid_number"), F.col("total_amount_clean_str").cast("double")).otherwise(F.lit(None).cast("double"))
)
df = df.withColumn(
"total_amount_num_dec",
F.when(F.col("total_amount_num_dbl").isNotNull(), F.round(F.col("total_amount_num_dbl"), 2).cast(DECIMAL_TYPE)).otherwise(F.lit(None).cast(DECIMAL_TYPE))
)
return df
def validate_total_amount(df_raw: DataFrame,
total_col: str = "total_amount",
sample_fraction: float = 0.1,
seed: int = 42) -> Dict[str, DataFrame]:
"""
Apply validation rules on fact_order_item.total_amount.
Returns dict with:
- valid_df: records passing all checks, total_amount normalized to DECIMAL(18,2) in column total_amount_std
- errors_df: row-level error details (multiple rows per input possible if multiple failures)
- consistency_conflicts_df: group-level inconsistencies across partitions/batches
- diff_report_df: summary stats for recomputation differences
- dist_stats_df: distribution stats for spike detection
"""
df = normalize_total_amount(df_raw, total_col)
# Non-empty check
df = df.withColumn("is_empty_str", F.trim(F.col("total_amount_raw_str")) == "")
# Non-null check
df = df.withColumn("is_null", F.col(total_col).isNull())
# Type check failures
type_fail = (F.col("invalid_number") | F.col("total_amount_num_dec").isNull())
# Range check: 0 <= amount <= 1_000_000
df = df.withColumn("range_underflow", F.col("total_amount_num_dec") < F.lit(0).cast(DECIMAL_TYPE))
df = df.withColumn("range_overflow", F.col("total_amount_num_dec") > F.lit(AMOUNT_UPPER_LIMIT).cast(DECIMAL_TYPE))
# Recompute expected amount
q = F.col("quantity").cast("double") if "quantity" in df.columns else F.lit(None).cast("double")
p = F.col("unit_price").cast("double") if "unit_price" in df.columns else F.lit(None).cast("double")
has_disc = "discount_amount" in df.columns
has_tax = "tax_amount" in df.columns
d = F.coalesce(F.col("discount_amount").cast("double"), F.lit(0.0)) if has_disc else F.lit(0.0)
t = F.coalesce(F.col("tax_amount").cast("double"), F.lit(0.0)) if has_tax else F.lit(0.0)
base = q * p
expected_expr = F.when(
((F.col("discount_amount").isNull()) & (F.col("tax_amount").isNull())) if (has_disc and has_tax) else F.lit(False),
base
).otherwise(base - d + t)
# Only compute when q and p are present
expected_expr = F.when(q.isNotNull() & p.isNotNull(), expected_expr).otherwise(F.lit(None).cast("double"))
df = df.withColumn("expected_total_dbl", expected_expr)
df = df.withColumn("expected_total_dec", F.when(F.col("expected_total_dbl").isNotNull(),
F.round(F.col("expected_total_dbl"), 2).cast(DECIMAL_TYPE)).otherwise(F.lit(None).cast(DECIMAL_TYPE)))
# Absolute difference in double for tolerance comparison
df = df.withColumn("recalc_abs_diff",
F.when(F.col("expected_total_dec").isNotNull() & F.col("total_amount_num_dec").isNotNull(),
F.abs(F.col("total_amount_num_dec").cast("double") - F.col("expected_total_dec").cast("double")))
.otherwise(F.lit(None).cast("double")))
df = df.withColumn("recalc_mismatch",
F.when(F.col("recalc_abs_diff").isNotNull(), F.col("recalc_abs_diff") > F.lit(AMOUNT_TOLERANCE))
.otherwise(F.lit(False)))
# Sample-based verification (reduces cost in batch). Records not in sample keep `recalc_mismatch` as computed (can be skipped if None).
df_sample = df.sample(withReplacement=False, fraction=sample_fraction, seed=seed)
df = (df.alias("all")
.join(df_sample.select("order_id").alias("s"), on=["order_id"], how="left") # sample join key by order for better coverage
.withColumn("in_sample", F.col("s.order_id").isNotNull())
.drop("s.order_id"))
# For rows outside sample, we can optionally null out mismatch to not flag them in batch; here we keep full computation
# If you prefer sampling strictly, uncomment the next line:
# df = df.withColumn("recalc_mismatch", F.when(F.col("in_sample"), F.col("recalc_mismatch")).otherwise(F.lit(False)))
# Consistency check across batches/partitions for same natural key
# Define natural key columns
key_cols = []
for candidates in [["order_id", "order_item_id"], ["order_id", "item_id"], ["order_id", "sku_id"]]:
if has_cols(df, candidates):
key_cols = candidates
break
if not key_cols:
key_cols = ["order_id"] # fallback if item-level key not present
# Use available batch/partition identifiers for diagnostics
part_cols = [c for c in ["batch_id", "ingest_date", "etl_date", "dt", "partition_date"] if c in df.columns]
# Build error rows
error_rows = []
# Non-null and empty checks
nonnull_fail_df = df.where(F.col("is_null") | F.col("is_empty_str")) \
.select(*(key_cols + part_cols),
F.col("total_amount_raw_str").alias("value"),
F.lit("NON_NULL").alias("error_type"),
F.when(F.col("is_null"), F.lit("total_amount is NULL"))
.when(F.col("is_empty_str"), F.lit("total_amount is empty string"))
.otherwise(F.lit("unknown")).alias("error_reason"))
error_rows.append(nonnull_fail_df)
# Type failures (invalid_number or cast failed)
type_fail_df = df.where(type_fail) \
.select(*(key_cols + part_cols),
F.col("total_amount_raw_str").alias("value"),
F.lit("TYPE_MISMATCH").alias("error_type"),
F.when(F.col("sci_notation_found"), F.lit("scientific notation forbidden"))
.when(F.col("invalid_number"), F.lit("non-numeric after cleaning"))
.otherwise(F.lit("cast to DECIMAL(18,2) failed")).alias("error_reason"))
error_rows.append(type_fail_df)
# Range failures
range_fail_df = df.where(F.col("range_underflow") | F.col("range_overflow")) \
.select(*(key_cols + part_cols),
F.col("total_amount_raw_str").alias("value"),
F.lit("RANGE").alias("error_type"),
F.when(F.col("range_underflow"), F.lit("amount < 0 not allowed"))
.when(F.col("range_overflow"), F.lit("amount > 1,000,000 requires review"))
.otherwise(F.lit("range violation")).alias("error_reason"))
error_rows.append(range_fail_df)
# Recalculation mismatch (only when we have inputs)
recalc_fail_df = df.where(F.col("recalc_mismatch") & F.col("expected_total_dec").isNotNull()) \
.select(*(key_cols + part_cols),
F.col("total_amount_num_dec").cast("string").alias("value"),
F.lit("RECALC_MISMATCH").alias("error_type"),
F.format_string("expected=%s, actual=%s, diff=%.4f",
F.col("expected_total_dec").cast("string"),
F.col("total_amount_num_dec").cast("string"),
F.col("recalc_abs_diff")).alias("error_reason"))
error_rows.append(recalc_fail_df)
errors_df = None
if error_rows:
errors_df = error_rows[0]
for e in error_rows[1:]:
errors_df = errors_df.unionByName(e, allowMissingColumns=True)
# Consistency conflicts: same key has multiple distinct amounts across partitions/batches
# We check distinct DECIMAL(18,2) representations
cons_group = (df.where(F.col("total_amount_num_dec").isNotNull())
.groupBy(*key_cols)
.agg(F.countDistinct("total_amount_num_dec").alias("distinct_amounts"),
F.collect_set("total_amount_num_dec").alias("amount_set"),
*([F.collect_set(c).alias(f"{c}_set") for c in part_cols] if part_cols else []))
.where(F.col("distinct_amounts") > 1))
consistency_conflicts_df = cons_group
# Distribution stats for spike detection
# Use approx_percentile; if not supported, fallback to approxQuantile via action
quantiles = [0.5, 0.9, 0.95, 0.99]
dist_stats_df = (df.where(F.col("total_amount_num_dec").isNotNull())
.agg(F.count("*").alias("n"),
F.min("total_amount_num_dec").alias("min"),
F.expr("percentile_approx(total_amount_num_dec, 0.5)").alias("p50"),
F.expr("percentile_approx(total_amount_num_dec, 0.9)").alias("p90"),
F.expr("percentile_approx(total_amount_num_dec, 0.95)").alias("p95"),
F.expr("percentile_approx(total_amount_num_dec, 0.99)").alias("p99"),
F.max("total_amount_num_dec").alias("max")))
# Diff report (summary) for recomputation mismatches within sample scope
diff_report_df = (df.where(F.col("expected_total_dec").isNotNull())
.agg(
F.sum(F.when(F.col("recalc_mismatch"), 1).otherwise(0)).alias("mismatch_cnt"),
F.count("*").alias("checked_cnt"),
F.avg(F.when(F.col("recalc_abs_diff").isNotNull(), F.col("recalc_abs_diff"))).alias("avg_abs_diff"),
F.max("recalc_abs_diff").alias("max_abs_diff")
))
# Valid records: pass all checks
valid_df = (df.where(
(~F.col("is_null")) &
(~F.col("is_empty_str")) &
(~type_fail) &
(~F.col("range_underflow")) &
(~F.col("range_overflow")) &
(~F.col("recalc_mismatch") | F.col("expected_total_dec").isNull()) # allow pass when expected not computable due to missing inputs
)
.withColumn("total_amount_std", F.col("total_amount_num_dec").cast(DECIMAL_TYPE))
# Optional: drop helper columns
.drop("total_amount_num_dbl", "total_amount_num_dec", "total_amount_clean_str",
"invalid_number", "numeric_like", "sci_notation_found",
"is_null", "is_empty_str", "range_underflow", "range_overflow",
"expected_total_dbl", "expected_total_dec", "recalc_abs_diff",
"recalc_mismatch", "in_sample", "total_amount_raw_str"))
return {
"valid_df": valid_df,
"errors_df": errors_df,
"consistency_conflicts_df": consistency_conflicts_df,
"diff_report_df": diff_report_df,
"dist_stats_df": dist_stats_df
}
-------------------------------
Example usage (adjust I/O as needed)
-------------------------------
result = validate_total_amount(df_raw, total_col="total_amount", sample_fraction=0.2, seed=7)
result["diff_report_df"].show(truncate=False)
result["dist_stats_df"].show(truncate=False)
