以下为一份可复用的 PySpark 数据质量验证脚本，针对所述线上贷款申请数据集，覆盖所要求的10类校验项。脚本具备配置化、可扩展和批处理/流式可切换的能力，并输出结构化的校验结果与样本明细。
脚本文件名建议：dq_loan_app_validation.py
依赖:
- Spark 3.x
- Python 3.8+
- 输入数据可为 parquet/csv/json（通过参数指定）
使用示例（批处理）:
spark-submit dq_loan_app_validation.py
--input /data/loan_apps/2024/
--input-format parquet
--output /data/dq_out/loan_apps/
--streaming false
代码:
#!/usr/bin/env python3
-- coding: utf-8 --
"""
贷款申请数据质量验证脚本
覆盖校验项：
- 缺失值检查
- 重复值检查
- 异常值检查（基于统计：IQR/极端分位）
- 异常值检查（基于业务规则：可配置）
- 数据类型一致性检查与统一
- 主键唯一性检查
- 值域范围检查（枚举/数值）
- 格式规范性检查（IP/哈希/时间）
- 逻辑一致性检查（跨字段）
- 时间序列连续性检查（按天/小时覆盖与缺口）
说明：
- 仅对存在于数据中的字段执行相应校验，缺失字段跳过并记录。
- 业务阈值使用可配置规则。默认值保守，建议依据历史数据调优。
- 结果以结构化明细与汇总指标方式输出，便于审计与复用。
"""
import argparse
import json
import sys
from datetime import datetime, timedelta
from pyspark.sql import SparkSession, DataFrame, Window
from pyspark.sql import functions as F
from pyspark.sql import types as T
-------------------------------
配置区（可按需调整/外部化为JSON/YAML）
-------------------------------
CONFIG = {
"primary_key": "application_id",
"timestamp_col": "app_datetime",
"id_type_col": "id_type",
"id_number_col": "id_number", # 已脱敏哈希
"core_numeric_cols": [
"age", "income_monthly", "debt_total", "loan_amount",
"loan_term_months", "employment_length_years", "credit_score"
],
"categorical_cols": [
"gender", "province", "marital_status", "education_level", "channel"
],
"string_cols": ["device_id", "ip"],
# 目标类型定义（统一类型）
"schema_expectation": {
"application_id": "string",
"app_datetime": "timestamp",
"id_type": "string",
"id_number": "string",
"age": "int",
"gender": "string",
"province": "string",
"income_monthly": "double",
"debt_total": "double",
"loan_amount": "double",
"loan_term_months": "int",
"employment_length_years": "double",
"marital_status": "string",
"education_level": "string",
"credit_score": "double",
"device_id": "string",
"ip": "string",
"channel": "string"
},
# 值域/枚举（如未知可留空，或仅做存在性检查）
"enums": {
"id_type": ["ID_CARD", "PASSPORT", "OTHER"],
"gender": ["M", "F", "U", "UNKNOWN", "OTHER"],
"marital_status": ["SINGLE", "MARRIED", "DIVORCED", "WIDOWED", "OTHER"],
"education_level": ["HIGH_SCHOOL", "BACHELOR", "MASTER", "PHD", "OTHER"],
"channel": [] # 若为空则不限制；否则精确匹配白名单
},
# 数值基本范围（保守默认，可根据历史数据调整；为None则跳过静态范围）
"ranges": {
"age": {"min": 18, "max": 75},
"loan_term_months": {"min": 1, "max": 120},
"employment_length_years": {"min": 0, "max": 60},
"income_monthly": {"min": 0, "max": None},
"debt_total": {"min": 0, "max": None},
"loan_amount": {"min": 0, "max": None},
"credit_score": {"min": 0, "max": None}
},
# 业务规则（可配置开关与阈值）
"business_rules": {
# 时间不超过未来5分钟，且不早于今天-400天（约13个月）
"app_datetime_future_minutes": 5,
"app_datetime_past_days": 400,
# DTI = debt_total/income_monthly：静态阈值与分位阈值取较宽松者
"max_dti_static": 5.0, # 若income=0则DTI设为null并记录异常
# 年龄与工作年限：age >= employment_length_years + 14（假设最早工作年龄14岁）
"min_age_minus_employment_years": 14,
# 重复申请：同(id_number, loan_amount, loan_term_months) 7天内重复视为重复
"repeat_within_days": 7
},
# 格式与规范
"format_checks": {
# IPv4正则
"ipv4_regex": r"^(?:(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d).){3}(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)$",
# 哈希长度白名单（常见：MD5=32, SHA-1=40, SHA-256=64），未知时可放宽
"id_hash_lengths": [32, 40, 64],
# device_id一般为字母数字与-，长度放宽
"device_id_regex": r"^[A-Za-z0-9-:.]{8,100}$",
# 允许的省份列表若为空不校验
"province_whitelist": []
},
# 统计异常检测（IQR）：乘数1.5为常规，3为极端
"iqr_multiplier": 1.5,
"extreme_iqr_multiplier": 3.0,
# 输出采样
"sample_rows": 100
}
-------------------------------
工具函数
-------------------------------
def spark_type_from_str(t: str):
m = {
"string": T.StringType(),
"int": T.IntegerType(),
"bigint": T.LongType(),
"double": T.DoubleType(),
"float": T.FloatType(),
"boolean": T.BooleanType(),
"timestamp": T.TimestampType(),
"date": T.DateType()
}
return m.get(t, T.StringType())
def ensure_columns(df: DataFrame, cols):
exists = [c for c in cols if c in df.columns]
missing = [c for c in cols if c not in df.columns]
return exists, missing
def cast_with_flag(df: DataFrame, col: str, to_type: str):
target_type = spark_type_from_str(to_type)
# 记录原始类型
cast_col = F.col(col).cast(target_type)
ok_col = F.when(F.col(col).isNull(), F.lit(True))
.when(cast_col.isNull() & F.col(col).isNotNull(), F.lit(False))
.otherwise(F.lit(True))
df2 = df.withColumn(col, cast_col)
.withColumn(f"{col}__cast_ok", ok_col)
return df2
def unify_types(df: DataFrame, schema_expectation: dict):
df2 = df
present = [c for c in schema_expectation.keys() if c in df.columns]
for c in present:
df2 = cast_with_flag(df2, c, schema_expectation[c])
return df2
def approx_iqr_bounds(df: DataFrame, col: str, multiplier: float):
q = df.approxQuantile(col, [0.25, 0.75], 0.001)
if not q or len(q) < 2:
return None
q1, q3 = q
if q1 is None or q3 is None:
return None
iqr = q3 - q1
lower = q1 - multiplier * iqr
upper = q3 + multiplier * iqr
return lower, upper
def write_df(df: DataFrame, path: str):
if df is None:
return
df.coalesce(1).write.mode("overwrite").parquet(path)
-------------------------------
校验逻辑
-------------------------------
def check_missingness(df: DataFrame, cfg: dict) -> DataFrame:
cols = [c for c in df.columns if not c.endswith("__cast_ok")]
exprs = [F.sum(F.col(c).isNull().cast("int")).alias(c) for c in cols]
miss = df.agg(*exprs)
miss_long = miss.select(F.explode(F.create_map([F.lit(x) for kv in sum(([F.lit(c), F.col(c)] for c in miss.columns), [])])).alias("col", "missing_count"))
miss_long = miss_long.withColumn("total_rows", F.lit(df.count()))
.withColumn("missing_rate", F.col("missing_count") / F.col("total_rows"))
return miss_long
def check_primary_key_uniqueness(df: DataFrame, pk: str) -> DataFrame:
if pk not in df.columns:
return None
dup = df.groupBy(pk).count().where(F.col("count") > 1)
return dup
def check_exact_duplicates(df: DataFrame) -> DataFrame:
# 完全相同行（全列相同）
grp_cols = df.columns
dup = df.groupBy(*grp_cols).count().where(F.col("count") > 1)
return dup
def check_semantic_duplicates(df: DataFrame, cfg: dict) -> DataFrame:
# 7天内重复申请：同(id_number, loan_amount, loan_term_months)
needed = ["id_number", "loan_amount", "loan_term_months", "app_datetime"]
exists, _ = ensure_columns(df, needed)
if len(exists) < 4:
return None
w = Window.partitionBy("id_number", "loan_amount", "loan_term_months").orderBy(F.col("app_datetime").cast("timestamp"))
prev_ts = F.lag("app_datetime").over(w)
seconds_diff = F.when(prev_ts.isNotNull(), F.col("app_datetime").cast("long") - prev_ts.cast("long")).otherwise(None)
within = df.withColumn("prev_app_datetime", prev_ts)
.withColumn("seconds_from_prev", seconds_diff)
.withColumn("repeat_within_days",
(F.col("seconds_from_prev").isNotNull()) &
(F.col("seconds_from_prev") <= F.lit(cfg["business_rules"]["repeat_within_days"] * 86400)))
return within.where(F.col("repeat_within_days") == True)
def check_type_consistency_flags(df: DataFrame) -> DataFrame:
cast_cols = [c for c in df.columns if c.endswith("__cast_ok")]
if not cast_cols:
return None
bad = []
for c in cast_cols:
src = c.replace("__cast_ok", "")
bad.append(F.sum((~F.col(c)).cast("int")).alias(src))
out = df.agg(*bad)
out_long = out.select(F.explode(F.create_map([F.lit(x) for kv in sum(([F.lit(c), F.col(c)] for c in out.columns), [])])).alias("col", "failed_casts"))
return out_long
def check_value_ranges(df: DataFrame, ranges: dict) -> DataFrame:
violations = None
for col, r in ranges.items():
if col not in df.columns:
continue
conds = []
if r.get("min") is not None:
conds.append(F.col(col) < F.lit(r["min"]))
if r.get("max") is not None:
conds.append(F.col(col) > F.lit(r["max"]))
if not conds:
continue
v = df.where(conds[0] if len(conds) == 1 else (conds[0] | conds[1]))
.select(F.lit(col).alias("col"), F.col(col).alias("value"), *[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
violations = v if violations is None else violations.unionByName(v)
return violations
def check_enums(df: DataFrame, enums: dict) -> DataFrame:
violations = None
for col, allowed in enums.items():
if col not in df.columns or not allowed:
continue
v = df.where(~F.col(col).isin(allowed) & F.col(col).isNotNull())
.select(F.lit(col).alias("col"), F.col(col).alias("value"), *[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
violations = v if violations is None else violations.unionByName(v)
return violations
def check_formats(df: DataFrame, cfg: dict) -> DataFrame:
viols = None
# IP
if "ip" in df.columns:
v1 = df.where(F.col("ip").isNotNull() & (~F.col("ip").rlike(cfg["format_checks"]["ipv4_regex"])))
.select(F.lit("ip").alias("col"), F.col("ip").alias("value"), *[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v1 if viols is None else viols.unionByName(v1)
# id_number哈希长度
if CONFIG["id_number_col"] in df.columns:
v2 = df.where(F.col(CONFIG["id_number_col"]).isNotNull() &
(~F.length(F.col(CONFIG["id_number_col"])).isin(cfg["format_checks"]["id_hash_lengths"])))
.select(F.lit(CONFIG["id_number_col"]).alias("col"), F.col(CONFIG["id_number_col"]).alias("value"), *[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v2 if viols is None else viols.unionByName(v2)
# device_id
if "device_id" in df.columns:
v3 = df.where(F.col("device_id").isNotNull() &
(~F.col("device_id").rlike(cfg["format_checks"]["device_id_regex"])))
.select(F.lit("device_id").alias("col"), F.col("device_id").alias("value"), *[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v3 if viols is None else viols.unionByName(v3)
# 省份白名单
if "province" in df.columns and cfg["format_checks"]["province_whitelist"]:
v4 = df.where(F.col("province").isNotNull() & (~F.col("province").isin(cfg["format_checks"]["province_whitelist"])))
.select(F.lit("province").alias("col"), F.col("province").alias("value"), *[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v4 if viols is None else viols.unionByName(v4)
return viols
def check_business_rules(df: DataFrame, cfg: dict) -> DataFrame:
br = cfg["business_rules"]
viols = None
now = F.current_timestamp()
# 时间窗口：不在未来+5分钟，且不早于past_days
if CONFIG["timestamp_col"] in df.columns:
v1 = df.where(
(F.col(CONFIG["timestamp_col"]) > F.expr(f"timestampadd(MINUTE, {br['app_datetime_future_minutes']}, current_timestamp())")) |
(F.col(CONFIG["timestamp_col"]) < F.expr(f"timestampadd(DAY, -{br['app_datetime_past_days']}, current_timestamp())"))
).select(F.lit(CONFIG["timestamp_col"]).alias("rule"),
F.col(CONFIG["timestamp_col"]).alias("value"),
*[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v1 if viols is None else viols.unionByName(v1)
# DTI
if "debt_total" in df.columns and "income_monthly" in df.columns:
dti = df.withColumn("DTI",
F.when(F.col("income_monthly") > 0, F.col("debt_total") / F.col("income_monthly"))
.otherwise(F.lit(None)))
v2 = dti.where(F.col("DTI").isNotNull() & (F.col("DTI") > F.lit(br["max_dti_static"])))
.select(F.lit("DTI_gt_static").alias("rule"),
F.col("DTI").alias("value"),
*[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v2 if viols is None else viols.unionByName(v2)
# income=0但debt>0的异常
v2b = df.where((F.col("income_monthly") <= 0) & (F.col("debt_total") > 0))
.select(F.lit("income_zero_but_debt_positive").alias("rule"),
F.struct("income_monthly", "debt_total").alias("value"),
*[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v2b if viols is None else viols.unionByName(v2b)
# 年龄与工作年限
if "age" in df.columns and "employment_length_years" in df.columns:
v3 = df.where(F.col("age") < (F.col("employment_length_years") + F.lit(br["min_age_minus_employment_years"])))
.select(F.lit("age_vs_employment_years").alias("rule"),
F.struct("age", "employment_length_years").alias("value"),
*[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v3 if viols is None else viols.unionByName(v3)
# 贷款金额必须 > 0
if "loan_amount" in df.columns:
v4 = df.where(F.col("loan_amount") <= 0)
.select(F.lit("loan_amount_positive").alias("rule"),
F.col("loan_amount").alias("value"),
*[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
viols = v4 if viols is None else viols.unionByName(v4)
return viols
def check_statistical_outliers(df: DataFrame, cfg: dict) -> DataFrame:
violations = None
for col in cfg["core_numeric_cols"]:
if col not in df.columns:
continue
bounds = approx_iqr_bounds(df.where(F.col(col).isNotNull()), col, cfg["iqr_multiplier"])
if not bounds:
continue
lower, upper = bounds
v = df.where((F.col(col) < F.lit(lower)) | (F.col(col) > F.lit(upper)))
.select(F.lit(col).alias("col"),
F.col(col).alias("value"),
F.lit(lower).alias("iqr_lower"),
F.lit(upper).alias("iqr_upper"),
*[F.col(c) for c in [CONFIG["primary_key"]] if c in df.columns])
violations = v if violations is None else violations.unionByName(v)
return violations
def check_time_continuity(df: DataFrame, cfg: dict) -> (DataFrame, DataFrame):
# 连续性检查：按天/小时统计并找出缺口
ts = CONFIG["timestamp_col"]
if ts not in df.columns:
return None, None
# 聚合
daily = df.groupBy(F.to_date(F.col(ts)).alias("date")).count()
hourly = df.groupBy(F.date_format(F.col(ts), "yyyy-MM-dd HH:00:00").alias("hour")).count()
# 生成日期序列
min_max = df.agg(F.min(F.to_date(F.col(ts))).alias("min_d"),
F.max(F.to_date(F.col(ts))).alias("max_d")).collect()[0]
if min_max["min_d"] is None or min_max["max_d"] is None:
return daily, hourly
min_d = min_max["min_d"]
max_d = min_max["max_d"]
spark = df.sql_ctx.sparkSession
dates = spark.range(0, F.datediff(F.lit(max_d), F.lit(min_d)) + 1)\
.withColumn("date", F.expr(f"date_add(to_date('{min_d}'), cast(id as int))"))\
.select("date")
daily_full = dates.join(daily, on="date", how="left").fillna({"count": 0})
missing_days = daily_full.where(F.col("count") == 0)
# 小时级序列（窗口按整点）
# 计算小时范围
min_max_h = df.agg(F.min(F.date_trunc("hour", F.col(ts))).alias("min_h"),
F.max(F.date_trunc("hour", F.col(ts))).alias("max_h")).collect()[0]
if min_max_h["min_h"] is None or min_max_h["max_h"] is None:
return daily_full, hourly
min_h = min_max_h["min_h"]
max_h = min_max_h["max_h"]
hours = spark.range(0, (int((max_h.timestamp() - min_h.timestamp()) // 3600) + 1))\
.withColumn("hour", F.expr(f"timestampadd(HOUR, cast(id as int), timestamp('{min_h}'))"))\
.select("hour")
hourly_full = hours.join(hourly.withColumn("hour", F.col("hour").cast("timestamp")), on="hour", how="left").fillna({"count": 0})
missing_hours = hourly_full.where(F.col("count") == 0)
# 返回覆盖与缺口
return missing_days, missing_hours
-------------------------------
主流程
-------------------------------
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True, help="输入路径（目录或文件）")
parser.add_argument("--input-format", default="parquet", choices=["parquet", "csv", "json"], help="输入格式")
parser.add_argument("--output", required=True, help="输出目录")
parser.add_argument("--streaming", default="false", choices=["true", "false"], help="是否使用Structured Streaming")
parser.add_argument("--csv-header", default="true", choices=["true", "false"], help="CSV是否包含表头")
args = parser.parse_args()
spark = SparkSession.builder.appName("DQ_Loan_Applications").getOrCreate()
spark.sparkContext.setLogLevel("WARN")
# 读入
if args.streaming == "true":
# 流式模式（示例）：以文件夹为源，micro-batch执行；注意：某些汇总（如approxQuantile）需使用complete/append触发器或批流分离
if args.input_format == "parquet":
df = spark.readStream.format("parquet").load(args.input)
elif args.input_format == "csv":
df = spark.readStream.option("header", args.csv_header == "true").option("inferSchema", "true").csv(args.input)
else:
df = spark.readStream.json(args.input)
# 简化：流式仅做记录级别规则检查与落地（完整统计建议离线批处理）
print("注意：流式模式下统计型/IQR类校验建议离线批处理运行，本脚本流式仅演示结构，可在生产中分层实现。")
sys.exit(0)
else:
if args.input_format == "parquet":
df = spark.read.parquet(args.input)
elif args.input_format == "csv":
df = spark.read.option("header", args.csv_header == "true").option("inferSchema", "true").csv(args.input)
else:
df = spark.read.json(args.input)
# 统一类型并标记类型转换失败
df = unify_types(df, CONFIG["schema_expectation"])
# 1) 缺失值检查
missing_report = check_missingness(df, CONFIG)
write_df(missing_report, f"{args.output}/missing_report")
# 2) 重复值检查
pk_dups = check_primary_key_uniqueness(df, CONFIG["primary_key"])
write_df(pk_dups, f"{args.output}/primary_key_duplicates")
exact_dups = check_exact_duplicates(df)
write_df(exact_dups, f"{args.output}/exact_duplicates")
semantic_dups = check_semantic_duplicates(df, CONFIG)
write_df(semantic_dups, f"{args.output}/semantic_duplicates")
# 3) 统计异常（IQR）
stat_outliers = check_statistical_outliers(df, CONFIG)
write_df(stat_outliers, f"{args.output}/statistical_outliers")
# 4) 业务规则异常
biz_violations = check_business_rules(df, CONFIG)
write_df(biz_violations, f"{args.output}/business_rule_violations")
# 5) 类型一致性检查（转换失败计数）
type_cast_report = check_type_consistency_flags(df)
write_df(type_cast_report, f"{args.output}/type_cast_report")
# 6) 主键唯一性（已在2中输出详细重复明细），这里输出唯一性状态
if pk_dups is not None:
pk_dup_count = pk_dups.count()
else:
pk_dup_count = None
# 7) 值域范围检查（数值范围）
range_violations = check_value_ranges(df, CONFIG["ranges"])
write_df(range_violations, f"{args.output}/range_violations")
# 8) 格式规范性检查
format_violations = check_formats(df, CONFIG)
write_df(format_violations, f"{args.output}/format_violations")
# 9) 逻辑一致性检查（在业务规则中覆盖了跨字段主要逻辑）
# 若需额外逻辑可在check_business_rules中扩展
# 10) 时间序列连续性检查
missing_days, missing_hours = check_time_continuity(df, CONFIG)
write_df(missing_days, f"{args.output}/missing_days")
write_df(missing_hours, f"{args.output}/missing_hours")
# 构建汇总指标
total_rows = df.count()
summary_rows = []
def cnt(df_):
return None if df_ is None else df_.count()
summary_rows.append(("total_rows", total_rows))
summary_rows.append(("missing_report_cols", cnt(missing_report)))
summary_rows.append(("primary_key_dup_count", cnt(pk_dups)))
summary_rows.append(("exact_duplicates_groups", cnt(exact_dups)))
summary_rows.append(("semantic_duplicates_rows", cnt(semantic_dups)))
summary_rows.append(("statistical_outliers_rows", cnt(stat_outliers)))
summary_rows.append(("business_rule_violations_rows", cnt(biz_violations)))
summary_rows.append(("type_cast_failed_cols", cnt(type_cast_report)))
summary_rows.append(("range_violations_rows", cnt(range_violations)))
summary_rows.append(("format_violations_rows", cnt(format_violations)))
summary_rows.append(("missing_days", cnt(missing_days)))
summary_rows.append(("missing_hours", cnt(missing_hours)))
summary_df = spark.createDataFrame(summary_rows, schema=["metric", "value"])
write_df(summary_df, f"{args.output}/summary_metrics")
# 导出各类异常样本（便于快速排查）
sample_targets = {
"statistical_outliers": stat_outliers,
"business_rule_violations": biz_violations,
"range_violations": range_violations,
"format_violations": format_violations,
"semantic_duplicates": semantic_dups,
"primary_key_duplicates": pk_dups
}
for name, d in sample_targets.items():
if d is not None and d.count() > 0:
write_df(d.limit(CONFIG["sample_rows"]), f"{args.output}/samples/{name}")
print("DQ校验完成。汇总指标：")
summary_df.show(truncate=False)
spark.stop()
if name == "main":
main()
实施与扩展说明:
- 配置外置：建议将 CONFIG 外置为JSON/YAML文件，通过命令行参数加载，支持不同渠道/批次的规则差异化。
- 渠道完备度：可在缺失值检查基础上，追加按channel的完备度透视（groupBy('channel').agg(missing率))，用于定位渠道特异性问题。
- 风险特征分布漂移：添加按月/渠道的分布统计并与基线分布比较，以识别异常输入突变（非本脚本必需，推荐在数据监控层实现）。
- 流式适配：对Structured Streaming的场景，建议在ODS层清洗（类型统一/格式校验）并落地，再由离线批处理执行统计型/IQR/连续性等全量校验，保证结果稳定与可审计性。
- 指标落库：summary_metrics与各类违规明细可写入Hive/Delta，定期生成日报/告警。
