当前位置：首页 > article >正文

【使用Apache Flink 实现滑动窗口流式计算】

article 2025/1/30 13:41:54

什么是Flink？

Apache Flink是一个用于分布式流式处理和批处理的开源实时计算引擎。它具备低延迟、高吞吐量和 exactly-once 语义的特点，适用于各种实时数据处理场景。

Flink的核心概念

作业（Job）：Flink程序的执行单元。
数据流（DataStream）：表示连续的数据流，可以进行转换和计算。
窗口（Window）：用于对无限数据流进行有界的数据切片处理。
状态（State）：用于保存和管理中间计算结果。
时间语义（Event Time、Processing Time、Ingestion Time）：用于确定事件发生的时间。

我在 实时攻击行为分析模块 中，使用 Apache Flink 的滑动窗口流式计算如下：

1. 滑动窗口定义与数据源接入

数据源：
- 攻防引擎产生的攻击日志通过 Kafka 实时推送，每条日志包含字段：攻击时间戳、攻击类型（如 SQL 注入、XSS）、攻击结果（成功/失败）、目标 IP、攻击者 ID 等。
- Flink 通过 FlinkKafkaConsumer 订阅 Kafka 的 attack-log Topic，反序列化为 AttackEvent 对象。
窗口配置：
- 窗口类型：滑动窗口（Sliding Window），窗口大小 5秒，滑动间隔 1秒（每 1 秒输出一次过去 5 秒的统计结果）。
- 时间语义：采用 Event Time（以攻击日志中的时间戳为准），通过 Watermark 处理乱序事件（允许最大延迟 2 秒）。

DataStream<AttackEvent> attackStream = env  
    .addSource(new FlinkKafkaConsumer<>("attack-log", new AttackEventDeserializer(), properties))  
    .assignTimestampsAndWatermarks(WatermarkStrategy  
        .<AttackEvent>forBoundedOutOfOrderness(Duration.ofSeconds(2))  
        .withTimestampAssigner((event, timestamp) -> event.getTimestamp()));  

// 按攻击类型分组，开滑动窗口  
DataStream<AttackStat> windowedStat = attackStream  
    .keyBy(AttackEvent::getAttackType)  
    .window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))  
    .aggregate(new AttackStatAggregator());

2. 攻击成功率计算逻辑

聚合逻辑：
- 在窗口内统计每种攻击类型的 成功次数 与 总次数，计算成功率。
- 自定义 AggregateFunction 实现累加器（保存中间状态）与合并逻辑：

public class AttackStatAggregator implements AggregateFunction<AttackEvent, AttackStatAccumulator, AttackStat> {  
    @Override  
    public AttackStatAccumulator createAccumulator() {  
        return new AttackStatAccumulator(0, 0);  
    }  

    @Override  
    public AttackStatAccumulator add(AttackEvent event, AttackStatAccumulator acc) {  
        acc.totalCount++;  
        if (event.isSuccess()) acc.successCount++;  
        return acc;  
    }  

    @Override  
    public AttackStat getResult(AttackStatAccumulator acc) {  
        double successRate = (acc.totalCount == 0) ? 0 : (double) acc.successCount / acc.totalCount;  
        return new AttackStat(acc.totalCount, successRate);  
    }  

    @Override  
    public AttackStatAccumulator merge(AttackStatAccumulator a, AttackStatAccumulator b) {  
        return new AttackStatAccumulator(a.totalCount + b.totalCount, a.successCount + b.successCount);  
    }  
}

3. 实时指标存储与可视化

写入 Redis 排行榜：

攻击成功率计算结果通过 RedisSink 写入 Redis Sorted Set，按成功率从高到低排序，供前端实时展示 Top 10 攻击类型：

// Flink 输出到 Redis  
DataStream<AttackStat> windowedStat = ...;  
windowedStat.addSink(new RedisSink<>(  
    "attack-success-rate",  
    (stat, jedis) -> jedis.zadd("attack_rank", stat.getSuccessRate(), stat.getAttackType())  
));

持久化到 Apache Doris：

原始攻击日志与统计结果通过 Flink JDBC Connector 写入 Doris 的 attack_log 表与 attack_stat 表，支持 OLAP 分析（如按时间维度聚合攻击趋势）。

Doris 表设计示例：

-- 攻击日志表（明细数据）  
CREATE TABLE attack_log (  
    event_time DATETIME,  
    attack_type VARCHAR(32),  
    is_success BOOLEAN,  
    src_ip VARCHAR(15),  
    target_ip VARCHAR(15)  
) DUPLICATE KEY(event_time)  
DISTRIBUTED BY HASH(src_ip) BUCKETS 10;  

-- 攻击统计表（聚合数据）  
CREATE TABLE attack_stat (  
    window_start DATETIME,  
    window_end DATETIME,  
    attack_type VARCHAR(32),  
    success_rate DOUBLE  
) UNIQUE KEY(window_start, attack_type)  
DISTRIBUTED BY HASH(window_start) BUCKETS 5;

可视化看板：
- 前端通过 ECharts 监听 WebSocket 服务，实时拉取 Redis 中的排行榜数据，并绘制热力图（基于攻击目标 IP 的地理位置分布）。

4. 性能优化与容错机制

Flink 配置优化：
- 并行度调优：根据 Kafka Topic 分区数设置 Flink 并行度（如 8 并行度），避免数据倾斜。
- 状态后端：使用 RocksDB 作为状态后端，支持大状态持久化与故障恢复。
- Checkpoint 配置：每 30 秒触发一次 Checkpoint，启用 Exactly-Once 语义。
Redis 写入优化：
- 使用 Pipeline 批量写入，减少网络开销，提升吞吐量。
- 设置 Key 的 TTL（如 1 小时），避免内存无限增长。

5. 实际挑战与解决方案

问题 1：窗口延迟导致数据不完整

现象：部分攻击日志因网络延迟超过 Watermark 允许的 2 秒，导致窗口关闭后数据被丢弃。

解决方案：

开启 Flink 的 Allowed Lateness（允许额外延迟 5 秒），并注册 SideOutput 捕获迟到数据，补充更新统计结果：

OutputTag<AttackEvent> lateDataTag = new OutputTag<>("late-data"){};  
DataStream<AttackStat> windowedStat = attackStream  
    .keyBy(...)  
    .window(...)  
    .allowedLateness(Time.seconds(5))  
    .sideOutputLateData(lateDataTag)  
    .aggregate(...);  

// 处理迟到数据  
DataStream<AttackEvent> lateData = windowedStat.getSideOutput(lateDataTag);  
lateData.addSink(...); // 如写入 Doris 供离线补偿

问题 2：Doris 写入瓶颈
- 现象：高峰时段 Flink 批量写入 Doris 导致 CPU 占用率过高。
- 解决方案：
  - 调整 Doris 的 Batch Insert 参数（如 exec_mem_limit 和 load_parallelism）；
  - 启用 Stream Load 替代 JDBC，提升写入效率。