3. 数据模型设计：定义交易事件（TransactionEvent）POJO 类

在实时风控系统中，数据模型是规则引擎与 Flink CEP 模式匹配的基石。一个设计良好的 TransactionEvent POJO 类，需要同时满足 序列化效率、字段可扩展性 以及 Flink 状态管理兼容性。本节将定义核心字段、类型约束，并给出完整的 Java 实现。

3.1 核心字段定义

交易事件必须包含风控规则所需的最小特征集。下表列出了 TransactionEvent 的必选字段及其业务含义：

3.2 POJO 类实现（Java）

以下代码实现了 TransactionEvent 类，包含全参构造器、无参构造器、Getter/Setter 方法，并重写了 toString() 便于日志输出。该类可直接用于 Flink DataStream 的 KeyBy 和 CEP 模式匹配。

import java.io.Serializable;
import java.util.Objects;

public class TransactionEvent implements Serializable {

    private static final long serialVersionUID = 1L;

    // ---------- 核心字段 ----------
    private String eventId;
    private String userId;
    private Double amount;
    private String currency;
    private Long timestamp;
    private String deviceId;
    private String ipAddress;
    private String merchantId;
    private Integer riskScore;

    // ---------- 构造器 ----------
    public TransactionEvent() {}

    public TransactionEvent(String eventId, String userId, Double amount,
                            String currency, Long timestamp, String deviceId,
                            String ipAddress, String merchantId, Integer riskScore) {
        this.eventId = eventId;
        this.userId = userId;
        this.amount = amount;
        this.currency = currency;
        this.timestamp = timestamp;
        this.deviceId = deviceId;
        this.ipAddress = ipAddress;
        this.merchantId = merchantId;
        this.riskScore = riskScore;
    }

    // ---------- Getter / Setter ----------
    public String getEventId() { return eventId; }
    public void setEventId(String eventId) { this.eventId = eventId; }

    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }

    public Double getAmount() { return amount; }
    public void setAmount(Double amount) { this.amount = amount; }

    public String getCurrency() { return currency; }
    public void setCurrency(String currency) { this.currency = currency; }

    public Long getTimestamp() { return timestamp; }
    public void setTimestamp(Long timestamp) { this.timestamp = timestamp; }

    public String getDeviceId() { return deviceId; }
    public void setDeviceId(String deviceId) { this.deviceId = deviceId; }

    public String getIpAddress() { return ipAddress; }
    public void setIpAddress(String ipAddress) { this.ipAddress = ipAddress; }

    public String getMerchantId() { return merchantId; }
    public void setMerchantId(String merchantId) { this.merchantId = merchantId; }

    public Integer getRiskScore() { return riskScore; }
    public void setRiskScore(Integer riskScore) { this.riskScore = riskScore; }

    // ---------- 工具方法 ----------
    @Override
    public String toString() {
        return "TransactionEvent{" +
                "eventId='" + eventId + '\'' +
                ", userId='" + userId + '\'' +
                ", amount=" + amount +
                ", currency='" + currency + '\'' +
                ", timestamp=" + timestamp +
                ", deviceId='" + deviceId + '\'' +
                ", ipAddress='" + ipAddress + '\'' +
                ", merchantId='" + merchantId + '\'' +
                ", riskScore=" + riskScore +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        TransactionEvent that = (TransactionEvent) o;
        return Objects.equals(eventId, that.eventId);
    }

    @Override
    public int hashCode() {
        return Objects.hash(eventId);
    }
}

3.3 数据模型与 Flink CEP 的映射关系

下图展示了 TransactionEvent 字段如何被 Flink CEP 模式匹配引擎消费，以及各字段在规则链中的流转路径：

3.4 字段扩展与版本兼容

生产环境中，交易事件字段会随业务迭代而增加。建议遵循以下扩展原则：

• 向后兼容： 新增字段必须设置默认值（如 null 或 0），避免反序列化时破坏已有作业状态。
• POJO 与 Avro 双模： 内部流转使用 Avro 序列化（压缩率高），外部接口保留 POJO 便于调试。
• 元数据字段： 预留 Map<String, String> extendInfo 字段，用于承载临时扩展属性（如渠道号、优惠券 ID），避免频繁修改 POJO 结构。