4. Spark SQL 数据清洗：空值处理、异常值过滤、数据去重、字段标准化

在催收数据管道中，原始数据通常来自多个上游系统（贷前审批、账务核心、外呼平台、第三方征信），存在大量空值、异常值、重复记录和字段格式不一致问题。本章聚焦 Spark SQL 在催收场景下的四项核心清洗操作，确保下游熔断策略和模型特征计算基于高质量数据。

4.1 空值处理：填充与过滤策略

催收数据中的空值主要出现在 逾期天数、联系方式、收入证明 等关键字段。Spark SQL 提供 fillna、dropna 和 COALESCE 三种方式处理空值。

-- 使用 COALESCE 实现多级联系方式回退
SELECT
  user_id,
  COALESCE(mobile_phone, home_phone, work_phone, '未知') AS contact_phone,
  COALESCE(overdue_days, 0) AS overdue_days
FROM raw_collection_data

4.2 异常值过滤：基于业务规则与统计方法

催收数据异常值通常表现为：逾期天数为负数、欠款金额超过授信额度 10 倍、年龄小于 18 或大于 100。Spark SQL 通过 WHERE 子句和 WHEN 条件表达式实现过滤。

-- 过滤异常值：逾期天数 >= 0 且欠款金额在合理范围
SELECT *
FROM cleaned_data
WHERE overdue_days >= 0
  AND overdue_days <= 180
  AND debt_amount BETWEEN 0 AND credit_limit * 5
  AND age BETWEEN 18 AND 70

4.3 数据去重：精确去重与模糊去重

催收系统常因重复推送、接口重试导致同一笔债务出现多条记录。去重策略分为 精确去重（基于主键）和 模糊去重（基于相似度）。

-- 精确去重：按合同号保留最新催收记录
SELECT contract_id, user_id, overdue_days, debt_amount, update_time
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY contract_id ORDER BY update_time DESC) AS rn
  FROM raw_collection_data
) tmp
WHERE rn = 1

4.4 字段标准化：统一格式与编码

不同上游系统对同一字段的表示方式各异，例如日期格式 yyyy-MM-dd vs yyyy/MM/dd，性别编码 M/F vs 1/0。标准化是后续聚合和特征工程的基础。

• 日期标准化： 统一为 yyyy-MM-dd 或时间戳
• 枚举值映射： 使用 CASE WHEN 或 CREATE_MAP 统一编码
• 字符串清洗： 去除首尾空格、全角转半角、统一大小写

-- 字段标准化示例
SELECT
  user_id,
  TO_DATE(date_str, 'yyyy/MM/dd') AS standard_date,
  CASE UPPER(TRIM(gender))
    WHEN 'M' THEN '男'
    WHEN 'F' THEN '女'
    WHEN '1' THEN '男'
    WHEN '0' THEN '女'
    ELSE '未知'
  END AS gender_std,
  LOWER(TRIM(email)) AS email_std
FROM raw_data

4.5 清洗流水线整合示例

将上述四个步骤组合为一个完整的 Spark SQL 清洗流水线，适用于每日催收数据批处理任务。

-- 完整清洗流水线
WITH step1_null AS (
  SELECT
    user_id,
    COALESCE(overdue_days, 0) AS overdue_days,
    COALESCE(debt_amount, 0) AS debt_amount,
    COALESCE(contact_phone, '未知') AS contact_phone
  FROM raw_collection_data
  WHERE id_card IS NOT NULL  -- 关键字段非空
),
step2_outlier AS (
  SELECT *
  FROM step1_null
  WHERE overdue_days >= 0 AND overdue_days <= 180
    AND debt_amount >= 0
),
step3_dedup AS (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY contract_id ORDER BY update_time DESC) AS rn
  FROM step2_outlier
),
step4_standard AS (
  SELECT
    user_id,
    contract_id,
    overdue_days,
    debt_amount,
    TO_DATE(update_time, 'yyyy-MM-dd HH:mm:ss') AS standard_date,
    LOWER(TRIM(contact_phone)) AS phone_std
  FROM step3_dedup
  WHERE rn = 1
)
SELECT * FROM step4_standard