3. 催收数据模型设计：客户画像表、逾期订单表、催收行为表、催收结果表

催收数据模型是催收系统的核心资产，它决定了催收策略的精准度、催收效率以及后续数据分析的深度。在 Spark 批流一体架构下，数据模型需要兼顾离线批处理的高吞吐与实时流处理的低延迟。本章围绕四个核心主题表展开：客户画像表、逾期订单表、催收行为表和催收结果表。

3.1 客户画像表 (customer_profile)

客户画像表是催收分析的起点，聚合了客户的基础信息、信用评分、历史行为标签和实时风险等级。该表以 customer_id 为主键，每天由 Spark 批处理任务全量刷新。

3.2 逾期订单表 (overdue_order)

逾期订单表记录每一笔订单的逾期状态变化，采用拉链表设计，支持时间旅行查询。每个订单从逾期开始到结清或核销，会生成多条记录，通过 effective_date 和 expiration_date 标记有效期。

-- 逾期订单拉链表 DDL 示例 (Spark SQL)
CREATE TABLE overdue_order (
    order_id          STRING,
    customer_id       STRING,
    loan_amount       DECIMAL(18,2),
    overdue_principal DECIMAL(18,2),
    overdue_days      INT,
    overdue_stage     STRING,  -- M1, M2, M3, M4+
    effective_date    DATE,
    expiration_date   DATE,
    is_current        BOOLEAN
)
USING parquet
PARTITIONED BY (dt STRING)
CLUSTERED BY (customer_id) INTO 128 BUCKETS;

3.3 催收行为表 (collection_action)

催收行为表是事件驱动的流水表，记录每一次催收触达行为，包括触达方式、时间、内容摘要和关联的催收员。该表由 Spark Structured Streaming 实时写入，同时供离线分析使用。

// Spark Structured Streaming 写入催收行为表示例
Dataset<Row> actions = spark
    .readStream()
    .format("kafka")
    .option("subscribe", "collection_actions")
    .load()
    .selectExpr(
        "CAST(value AS STRING) as json",
        "CAST(timestamp AS TIMESTAMP) as action_time"
    )
    .select(
        functions.from_json(
            functions.col("json"),
            actionSchema
        ).alias("data"),
        functions.col("action_time")
    )
    .select("data.*", "action_time")
    .writeStream()
    .format("parquet")
    .option("checkpointLocation", "/data/checkpoint/collection_action")
    .partitionBy("action_date")
    .start();

3.4 催收结果表 (collection_result)

催收结果表记录每一次催收行为的最终结果，包括承诺还款金额、实际还款金额、还款日期等。该表与催收行为表通过 action_id 关联，形成行为-结果一对一的闭环。

3.5 模型关联与 Spark 实践

四个核心表通过外键形成星型模型，客户画像表作为维度表，逾期订单表作为事实表，催收行为表和催收结果表作为明细事件表。在 Spark 中，典型的 ETL 流程如下：

1. 每日批处理：使用 Spark SQL 将客户画像表与逾期订单表 Join，生成催收任务宽表，写入催收行为表。
2. 实时流处理：Spark Structured Streaming 消费 Kafka 中的催收行为事件，写入催收行为表分区。
3. 结果回写：催收结果表通过异步消息队列更新，Spark 微批次作业每 5 分钟拉取结果，更新逾期订单表的 overdue_stage 和客户画像表的 internal_risk_level。

-- Spark SQL 催收任务宽表生成示例
CREATE OR REPLACE TEMP VIEW collection_task AS
SELECT
    p.customer_id,
    p.internal_risk_level,
    o.order_id,
    o.overdue_principal,
    o.overdue_days,
    o.overdue_stage,
    CURRENT_TIMESTAMP() AS task_create_time
FROM customer_profile p
INNER JOIN overdue_order o
    ON p.customer_id = o.customer_id
WHERE o.is_current = TRUE
  AND o.overdue_days > 0;