4、实时特征计算：Flink SQL与DataStream API实现滑动窗口计数、时间衰减统计、复杂事件处理

在风控实时链路中，特征计算是规则引擎与模型推理的前置核心环节。本章聚焦于三种高频实时特征模式：滑动窗口计数（如近5分钟交易次数）、时间衰减统计（如加权累计金额）以及复杂事件处理（如登录后未改密即交易）。我们将分别使用 Flink SQL 的声明式语法与 DataStream API 的底层算子进行实现，并对比其适用场景。

4.1 滑动窗口计数：Flink SQL vs DataStream API

滑动窗口（Sliding Window）是风控中最基础的统计模式，用于计算最近一段时间内的聚合指标。Flink SQL 通过 HOP 窗口函数实现，DataStream API 则通过 SlidingEventTimeWindows 或自定义 ProcessWindowFunction 完成。

4.1.1 Flink SQL 实现（HOP 窗口）

-- 近5分钟交易次数，每1分钟滑动一次
CREATE VIEW recent_trade_count AS
SELECT
    user_id,
    COUNT(*) AS trade_cnt_5m,
    HOP_START(event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) AS window_start,
    HOP_END(event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) AS window_end
FROM trade_event
GROUP BY
    user_id,
    HOP(event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE);

4.1.2 DataStream API 实现（ProcessWindowFunction）

DataStream<TradeEvent> tradeStream = ...;

DataStream<WindowResult> resultStream = tradeStream
    .keyBy(TradeEvent::getUserId)
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
    .process(new ProcessWindowFunction<TradeEvent, WindowResult, String, TimeWindow>() {
        @Override
        public void process(String key, Context context,
                            Iterable<TradeEvent> elements, Collector<WindowResult> out) {
            long count = 0;
            for (TradeEvent e : elements) {
                count++;
            }
            out.collect(new WindowResult(key, count,
                context.window().getStart(), context.window().getEnd()));
        }
    });

4.2 时间衰减统计：加权累计与半衰期模型

时间衰减统计（Time Decay）赋予近期事件更高的权重，常用于计算“近30分钟加权交易金额”或“行为置信度评分”。Flink SQL 原生不支持衰减函数，需借助自定义 UDAF；DataStream API 则可通过 ValueState 与定时器灵活实现。

4.2.1 DataStream API 实现（半衰期衰减）

public class DecayAggregateFunction extends KeyedProcessFunction<String, TradeEvent, Double> {
    private ValueState<Double> decayedValueState;
    private ValueState<Long> lastTimestampState;
    private final double halfLife = 300_000; // 5分钟半衰期（毫秒）

    @Override
    public void open(Configuration parameters) {
        decayedValueState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("decayedValue", Types.DOUBLE));
        lastTimestampState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("lastTimestamp", Types.LONG));
    }

    @Override
    public void processElement(TradeEvent event, Context ctx, Collector<Double> out) throws Exception {
        Double currentValue = decayedValueState.value();
        Long lastTime = lastTimestampState.value();
        if (currentValue == null) {
            currentValue = 0.0;
            lastTime = event.getEventTime();
        }
        // 时间衰减因子：2^(-Δt / halfLife)
        double decayFactor = Math.pow(2, -(event.getEventTime() - lastTime) / halfLife);
        double newValue = currentValue * decayFactor + event.getAmount();
        decayedValueState.update(newValue);
        lastTimestampState.update(event.getEventTime());
        out.collect(newValue);
    }
}

4.2.2 Flink SQL 自定义 UDAF（衰减求和）

public class DecaySumUDAF extends AggregateFunction<Double, DecayAccumulator> {
    @Override
    public DecayAccumulator createAccumulator() {
        return new DecayAccumulator();
    }

    public void accumulate(DecayAccumulator acc, Double amount, Long eventTime) {
        // 简化实现：仅保留最近一次事件的时间戳与加权值
        if (acc.lastTimestamp == null) {
            acc.lastTimestamp = eventTime;
            acc.weightedSum = amount;
        } else {
            double decay = Math.pow(2, -(eventTime - acc.lastTimestamp) / 300_000.0);
            acc.weightedSum = acc.weightedSum * decay + amount;
            acc.lastTimestamp = eventTime;
        }
    }

    @Override
    public Double getValue(DecayAccumulator acc) {
        return acc.weightedSum;
    }
}

4.3 复杂事件处理（CEP）：登录-交易序列检测

复杂事件处理（CEP）用于识别跨事件类型的时序模式，例如“同一设备在5分钟内登录失败3次后触发交易”。Flink 提供 CEP 库，DataStream API 可直接调用；Flink SQL 则通过 MATCH_RECOGNIZE 子句实现。

4.3.1 DataStream API + CEP 实现

Pattern<RiskEvent, ?> pattern = Pattern
    .<RiskEvent>begin("login_fail")
        .where(new SimpleCondition<RiskEvent>() {
            @Override
            public boolean filter(RiskEvent event) {
                return "LOGIN".equals(event.getEventType())
                    && "FAIL".equals(event.getResult());
            }
        })
        .times(3)
        .next("trade")
        .where(new SimpleCondition<RiskEvent>() {
            @Override
            public boolean filter(RiskEvent event) {
                return "TRADE".equals(event.getEventType());
            }
        })
        .within(Time.minutes(5));

PatternStream<RiskEvent> patternStream = CEP.pattern(
    riskStream.keyBy(RiskEvent::getDeviceId), pattern);

DataStream<Alert> alertStream = patternStream.process(
    new PatternProcessFunction<RiskEvent, Alert>() {
        @Override
        public void processMatch(
            Map<String, List<RiskEvent>> match, Context ctx, Collector<Alert> out) {
            out.collect(new Alert("HIGH_RISK_SEQUENCE", match));
        }
    });

4.3.2 Flink SQL MATCH_RECOGNIZE 实现

SELECT *
FROM risk_event
    MATCH_RECOGNIZE (
        PARTITION BY device_id
        ORDER BY event_time
        MEASURES
            A.event_time AS login_fail_start,
            B.event_time AS trade_time
        ONE ROW PER MATCH
        PATTERN (A{3,} B) WITHIN INTERVAL '5' MINUTE
        DEFINE
            A AS event_type = 'LOGIN' AND result = 'FAIL',
            B AS event_type = 'TRADE'
    ) AS matched;

4.4 三种特征计算模式对比

4.5 实时特征计算架构总览

下图展示了从事件接入到特征输出的完整链路，以及 Flink SQL 与 DataStream API 在其中的协作关系。

4.6 实践要点与避坑指南

• 状态大小控制：滑动窗口与衰减统计均需维护状态。建议为 ValueState 设置 TTL（Time-To-Live），避免状态无限膨胀。
• 事件时间语义：风控场景必须使用 EventTime，并配置合理的 Watermark 延迟（通常为 5-10 秒）。
• CEP 模式超时：使用 .within(Time.minutes(5)) 约束模式匹配窗口，防止长时间占用内存。
• SQL 与 API 混合：建议将简单聚合（如计数、求和）用 SQL 实现，复杂序列逻辑用 DataStream API + CEP 实现，通过 DataStream<-->Table 互转桥接。