2. Spark在催收中的应用：批处理与流处理、ETL与报表生成

催收系统每天需要处理数千万条逾期记录、通话记录、还款流水和外部征信数据。传统单机数据库在数据量超过TB级时，ETL耗时过长，报表生成延迟超过24小时，无法支撑实时催收策略调整。Spark凭借其内存计算、统一批流引擎和丰富的SQL/DataFrame API，成为催收数据中台的核心计算引擎。

2.1 催收数据ETL：从多源异构到统一宽表

催收数据源通常包括：核心账务系统的逾期明细、催收系统的案件分配记录、通话录音元数据、短信/邮件触达日志、第三方征信报告、以及实时还款流水。ETL的目标是将这些分散、异构的数据清洗、转换、关联后，形成一张催收宽表（Collection Wide Table），供后续报表和模型使用。

2.2 Spark批处理与流处理：统一引擎下的双模架构

催收场景天然需要批流一体：历史数据回测、月度报表生成使用批处理；实时案件分配、催收策略动态调整使用流处理。Spark从2.3开始，Structured Streaming实现了批流统一的DataFrame API，同一套业务逻辑可以无缝切换批/流模式。

// 批处理：读取Hive分区表，生成催收日报
val dailyReport = spark
  .table("collection_wide_table")
  .filter($"dt" === yesterday)
  .groupBy("collector_id", "product_type")
  .agg(
    count("*").as("total_cases"),
    sum(when($"status" === "paid", 1).otherwise(0)).as("paid_cases"),
    avg($"overdue_days").as("avg_overdue_days")
  )
dailyReport.write.mode("overwrite").saveAsTable("collection_daily_report")

// 流处理：实时监听还款事件，更新案件状态
val repaymentStream = spark
  .readStream
  .format("kafka")
  .option("subscribe", "repayment_topic")
  .load()
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json($"json", repaymentSchema).as("data"))
  .select("data.case_id", "data.amount", "data.timestamp")

repaymentStream
  .withWatermark("timestamp", "10 minutes")
  .groupBy("case_id")
  .agg(sum("amount").as("total_repaid"))
  .writeStream
  .foreachBatch { (batchDF, batchId) =>
    batchDF.createOrReplaceTempView("repayment_batch")
    spark.sql("""
      MERGE INTO collection_cases c
      USING repayment_batch r
      ON c.case_id = r.case_id
      WHEN MATCHED THEN UPDATE SET c.repaid_amount = r.total_repaid
    """)
  }
  .start()

2.3 催收报表生成：分层聚合与预计算

催收报表分为三个层级：运营层（催收员个人绩效、日回款率）、管理层（各产品逾期率、委外回收率）、决策层（全量资产质量、催收策略ROI）。Spark通过分层聚合（Multi-Level Aggregation）和预计算（Pre-Aggregation）来满足不同时效性要求。

催收报表分层架构

催收日报生成示例（Spark SQL）

-- 催收日报：按催收员、产品类型统计关键指标
INSERT OVERWRITE TABLE collection_daily_report PARTITION(dt='2025-03-21')
SELECT
    collector_id,
    product_type,
    COUNT(DISTINCT case_id) AS total_cases,
    SUM(CASE WHEN status = 'assigned' THEN 1 ELSE 0 END) AS assigned_cases,
    SUM(CASE WHEN status = 'in_progress' THEN 1 ELSE 0 END) AS in_progress_cases,
    SUM(CASE WHEN status = 'paid' THEN 1 ELSE 0 END) AS paid_cases,
    ROUND(AVG(overdue_days), 1) AS avg_overdue_days,
    ROUND(SUM(paid_amount) / NULLIF(SUM(total_amount), 0) * 100, 2) AS recovery_rate,
    COUNT(CASE WHEN contact_count > 0 THEN 1 END) AS contacted_cases,
    ROUND(AVG(contact_duration_sec), 0) AS avg_contact_duration
FROM collection_wide_table
WHERE dt = '2025-03-21'
GROUP BY collector_id, product_type;

2.4 催收数据ETL实战：增量更新与历史回溯

催收ETL面临的核心挑战是增量与全量混合：每天需要增量更新当天数据，同时定期对历史全量数据进行回溯修正（如修复前一天的错误标签）。Spark通过Delta Lake或Apache Iceberg等湖仓格式，实现了ACID语义下的增量合并（Merge Into）和时间旅行（Time Travel）。

增量ETL流程步骤

1. 增量抽取：从Kafka消费当天的催收案件变更事件（CDC），或从源系统拉取增量CSV/Parquet文件。
2. 数据清洗：过滤异常记录（如overdue_days < 0）、格式转换（时间戳统一为UTC）、字段补全（关联客户基本信息）。
3. 增量合并：使用Delta Lake的MERGE语句，将增量数据与历史宽表合并，支持INSERT/UPDATE/DELETE。
4. 质量校验：比对增量前后的记录数、关键指标（如总逾期金额）是否在合理阈值内，触发告警或回滚。
5. 报表刷新：触发下游日报/周报的增量刷新，避免全量重算。

-- Delta Lake增量合并示例：催收宽表每日更新
MERGE INTO collection_wide_table AS target
USING (
  SELECT
    case_id,
    collector_id,
    product_type,
    overdue_days,
    status,
    paid_amount,
    contact_count,
    contact_duration_sec,
    current_timestamp() AS update_time
  FROM collection_incremental
  WHERE dt = '2025-03-21'
) AS source
ON target.case_id = source.case_id
WHEN MATCHED THEN UPDATE SET
  target.collector_id = source.collector_id,
  target.status = source.status,
  target.paid_amount = source.paid_amount,
  target.contact_count = source.contact_count,
  target.update_time = source.update_time
WHEN NOT MATCHED THEN INSERT *

2.5 催收报表性能优化：预聚合与物化视图

催收报表的查询模式通常是固定维度聚合（按催收员、产品、日期）。Spark通过预聚合（Pre-Aggregation）和物化视图（Materialized View），将高频查询的中间结果提前计算并存储，避免每次查询都扫描全量宽表。

-- 预聚合宽表构建示例：按催收员+产品+日期预聚合
INSERT OVERWRITE TABLE collection_pre_agg
PARTITION(dt)
SELECT
    collector_id,
    product_type,
    dt,
    COUNT(DISTINCT case_id) AS total_cases,
    SUM(paid_amount) AS total_paid,
    AVG(overdue_days) AS avg_overdue_days,
    COUNT(CASE WHEN contact_count > 0 THEN 1 END) AS contacted_cases
FROM collection_wide_table
GROUP BY collector_id, product_type, dt;

-- 对催收员ID字段应用Z-Order索引
OPTIMIZE collection_pre_agg
ZORDER BY (collector_id);