4. 商户批次数据模型设计：商户维度表、批次流水表、风险标签表的设计原则

在实时风险迁移诊断中，数据模型是 Flink 作业的“骨架”。商户维度表、批次流水表与风险标签表分别承载了静态属性、动态行为与诊断结论。设计不当会导致状态膨胀、回溯困难或标签错乱。本章从实时计算视角，给出三张核心表的设计原则与字段规范。

4.1 商户维度表：静态画像与分区策略

商户维度表存储不频繁变化的属性，如注册时间、行业分类、风控等级、历史迁移次数。在 Flink 中通常作为 广播流（BroadcastStream） 或 维表 Join（AsyncIO / LookupJoin） 使用。

4.2 批次流水表：事件溯源与时间语义

批次流水表是实时诊断的“事实表”，每行记录一次批次操作（创建、审核、放款、迁移）。Flink 作业基于该表进行窗口聚合、模式匹配与风险评分。

-- 批次流水表 DDL（Flink SQL 示例）
CREATE TABLE batch_flow (
    batch_id        STRING,
    merchant_id     STRING,
    event_type      STRING,      -- 'CREATE' | 'AUDIT' | 'LOAN' | 'MIGRATE'
    amount          DECIMAL(18,2),
    event_time      TIMESTAMP(3),
    proc_time      AS PROCTIME(),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'batch_flow',
    'format' = 'json',
    'properties.group.id' = 'risk_diagnosis_group'
);

4.3 风险标签表：结论持久化与回溯

风险标签表存储 Flink 作业的诊断结论，包括标签类型、置信度、生效时间与过期时间。该表既用于下游通知，也用于离线回溯分析。

-- 风险标签表输出 Sink（Flink SQL 示例）
CREATE TABLE risk_label_sink (
    merchant_id     STRING,
    batch_id        STRING,
    label_type      STRING,      -- 'HIGH_RISK_MIGRATE' | 'ABNORMAL_FLOW' | ...
    confidence      DOUBLE,      -- 0.0 ~ 1.0
    label_time      TIMESTAMP(3),
    expire_time     TIMESTAMP(3),
    PRIMARY KEY (merchant_id, batch_id, label_type) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'risk_label',
    'key.format' = 'json',
    'value.format' = 'json'
);

4.4 三表关系与实时数据流

下图展示了三张表在 Flink 实时诊断中的协作关系：商户维度表提供静态画像，批次流水表驱动事件处理，风险标签表输出诊断结论。

4.5 设计原则对比清单