Sliding Windows

1. 事件时间滑动窗口

input
    .keyBy(<key selector>)
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    //                                 ^^^^^^^^^^^^^^^^^  ^^^^^^^^^^^^^^^^^
    //                                 窗口大小(size)      滑动步长(slide)
    .<windowed transformation>(<window function>);

2. 处理时间滑动窗口

input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .<windowed transformation>(<window function>);

3. 带偏移量的处理时间滑动窗口

input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(
        Time.hours(12),   // 窗口大小：12小时
        Time.hours(1),    // 滑动步长：1小时
        Time.hours(-8)    // 偏移量：-8小时（时区调整）
    ))
    .<windowed transformation>(<window function>);

核心概念：滑动窗口（Sliding Window）

关键参数

SlidingEventTimeWindows.of(size, slide)
//                         ^^^^  ^^^^^
//                         窗口大小  滑动步长
  • size（窗口大小）：每个窗口的时间跨度
  • slide（滑动步长）：窗口滑动的间隔

滑动窗口 vs 滚动窗口

特性滚动窗口滑动窗口
重叠❌ 不重叠✅ 可重叠
数据归属每条数据属于1个窗口每条数据可属于多个窗口
窗口间隔窗口大小 = 滑动步长窗口大小 > 滑动步长（通常）
使用场景独立统计平滑趋势、移动平均

窗口划分示意图

情况1：窗口大小 = 10秒，滑动步长 = 5秒（窗口重叠）
时间轴：
0    5    10   15   20   25   30
|----|----|----|----|----|----|

窗口1: [0,  10)  |==========|
窗口2:      [5,  15)  |==========|
窗口3:           [10, 20)  |==========|
窗口4:                [15, 25)  |==========|
窗口5:                     [20, 30)  |==========|

特点：
- 窗口重叠50%
- 每5秒产生一个新窗口
- 一条数据可能属于2个窗口

数据分配示例：

时间戳为8秒的数据：
- 属于窗口1 [0, 10)   ✅
- 属于窗口2 [5, 15)   ✅
- 属于窗口3 [10, 20)  ❌

时间戳为12秒的数据：
- 属于窗口2 [5, 15)   ✅
- 属于窗口3 [10, 20)  ✅
- 属于窗口4 [15, 25)  ❌
情况2：窗口大小 = 10秒，滑动步长 = 10秒（等同于滚动窗口）
窗口1: [0,  10)  |==========|
窗口2:            [10, 20)  |==========|
窗口3:                       [20, 30)  |==========|

特点：
- 窗口不重叠
- 退化为滚动窗口
情况3：窗口大小 = 10秒，滑动步长 = 2秒（高度重叠）
窗口1: [0,  10)  |==========|
窗口2:   [2,  12)  |==========|
窗口3:     [4,  14)  |==========|
窗口4:       [6,  16)  |==========|
窗口5:         [8,  18)  |==========|

特点：
- 每2秒产生一个新窗口
- 一条数据属于5个窗口（10/2=5）
- 更平滑的趋势分析

详细示例解析

示例1：事件时间滑动窗口（10秒窗口，5秒滑动）

public class SlidingEventTimeWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 模拟数据：用户点击事件
        DataStream<Event> events = env.fromElements(
            new Event("user1", 1, 1000L),   // 1秒
            new Event("user1", 1, 3000L),   // 3秒
            new Event("user1", 1, 7000L),   // 7秒
            new Event("user1", 1, 11000L),  // 11秒
            new Event("user1", 1, 13000L)   // 13秒
        );
        
        // 分配时间戳和Watermark
        DataStream<Event> withTimestamps = events.assignTimestampsAndWatermarks(
            WatermarkStrategy.<Event>forMonotonousTimestamps()
                .withTimestampAssigner((event, ts) -> event.timestamp)
        );
        
        // 滑动窗口：窗口大小10秒，每5秒滑动一次
        withTimestamps
            .keyBy(event -> event.userId)
            .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
            .process(new ProcessWindowFunction<Event, String, String, TimeWindow>() {
                @Override
                public void process(
                    String userId,
                    Context ctx,
                    Iterable<Event> elements,
                    Collector<String> out
                ) {
                    int count = 0;
                    for (Event e : elements) count++;
                    
                    out.collect(String.format(
                        "User: %s, Window: [%d, %d), Count: %d",
                        userId,
                        ctx.window().getStart() / 1000,
                        ctx.window().getEnd() / 1000,
                        count
                    ));
                }
            })
            .print();
        
        env.execute("Sliding Event-Time Window");
    }
    
    static class Event {
        String userId;
        int clickCount;
        Long timestamp;
        
        Event(String userId, int clickCount, Long timestamp) {
            this.userId = userId;
            this.clickCount = clickCount;
            this.timestamp = timestamp;
        }
    }
}

/* 输出分析：
事件时间戳：1s, 3s, 7s, 11s, 13s

窗口[0, 10):   包含 1s, 3s, 7s          → Count: 3
窗口[5, 15):   包含 7s, 11s, 13s        → Count: 3
窗口[10, 20):  包含 11s, 13s            → Count: 2

注意：
- 7s的数据同时属于窗口[0,10)和窗口[5,15)
- 11s的数据同时属于窗口[5,15)和窗口[10,20)
*/

示例2：处理时间滑动窗口（实时统计）

public class SlidingProcessingTimeWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 模拟实时数据流
        DataStream<Tuple2<String, Integer>> stream = env
            .socketTextStream("localhost", 9999)
            .map(line -> {
                String[] parts = line.split(",");
                return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
            });
        
        // 滑动窗口：10秒窗口，每5秒统计一次
        stream
            .keyBy(tuple -> tuple.f0)
            .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
            .sum(1)
            .print();
        
        /* 应用场景：
        实时监控最近10秒的数据，每5秒更新一次统计结果
        
        时间线（处理时间）：
        00:00:00-00:00:10: 窗口1统计
        00:00:05-00:00:15: 窗口2统计（与窗口1重叠50%）
        00:00:10-00:00:20: 窗口3统计
        
        优点：更平滑的趋势展示
        */
        
        env.execute("Sliding Processing-Time Window");
    }
}

示例3：带偏移量的滑动窗口（跨时区统计）

public class SlidingWindowWithOffset {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 模拟订单数据
        DataStream<Order> orders = env.fromElements(
            new Order("product1", 100.0),
            new Order("product1", 200.0),
            new Order("product2", 300.0)
        );
        
        // 滑动窗口：12小时窗口，每1小时滑动，按北京时间（UTC-8）
        orders
            .keyBy(order -> order.productId)
            .window(SlidingProcessingTimeWindows.of(
                Time.hours(12),   // 窗口大小：12小时
                Time.hours(1),    // 滑动步长：每小时统计一次
                Time.hours(-8)    // 偏移量：调整到北京时间
            ))
            .aggregate(new AggregateFunction<Order, Tuple2<Double, Integer>, Double>() {
                public Tuple2<Double, Integer> createAccumulator() {
                    return new Tuple2<>(0.0, 0);
                }
                
                public Tuple2<Double, Integer> add(Order order, Tuple2<Double, Integer> acc) {
                    return new Tuple2<>(acc.f0 + order.amount, acc.f1 + 1);
                }
                
                public Double getResult(Tuple2<Double, Integer> acc) {
                    return acc.f1 == 0 ? 0.0 : acc.f0 / acc.f1;
                }
                
                public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, 
                                                     Tuple2<Double, Integer> b) {
                    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
                }
            })
            .print();
        
        /* 应用场景：
        每小时统计最近12小时的平均订单金额
        
        窗口示例（北京时间）：
        窗口1: [2025-01-01 00:00, 2025-01-01 12:00)
        窗口2: [2025-01-01 01:00, 2025-01-01 13:00)
        窗口3: [2025-01-01 02:00, 2025-01-01 14:00)
        ...
        
        每小时更新一次，展示12小时移动平均
        */
        
        env.execute("Sliding Window with Offset");
    }
    
    static class Order {
        String productId;
        Double amount;
        
        Order(String productId, Double amount) {
            this.productId = productId;
            this.amount = amount;
        }
    }
}

常见应用场景

场景1：移动平均值（Moving Average）

// 计算最近10秒的平均温度，每2秒更新一次
temperatures
    .keyBy(sensor -> sensor.sensorId)
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
    .aggregate(new AggregateFunction<Temperature, Tuple2<Double, Integer>, Double>() {
        public Tuple2<Double, Integer> createAccumulator() {
            return new Tuple2<>(0.0, 0);
        }
        
        public Tuple2<Double, Integer> add(Temperature temp, Tuple2<Double, Integer> acc) {
            return new Tuple2<>(acc.f0 + temp.value, acc.f1 + 1);
        }
        
        public Double getResult(Tuple2<Double, Integer> acc) {
            return acc.f1 == 0 ? 0.0 : acc.f0 / acc.f1;  // 平均值
        }
        
        public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, 
                                             Tuple2<Double, Integer> b) {
            return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
        }
    });

/* 优点：
1. 更平滑的趋势展示
2. 减少噪声影响
3. 更频繁的更新
*/

场景2：实时TopN（每分钟统计最近5分钟的热门商品）

// 每分钟更新一次最近5分钟的热门商品Top 10
clicks
    .keyBy(click -> click.productId)
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
    .aggregate(
        // 增量统计点击数
        new AggregateFunction<Click, Long, Tuple2<String, Long>>() {
            public Long createAccumulator() { return 0L; }
            public Long add(Click click, Long acc) { return acc + 1; }
            public Tuple2<String, Long> getResult(Long count) { 
                return new Tuple2<>(getCurrentProductId(), count); 
            }
            public Long merge(Long a, Long b) { return a + b; }
        }
    )
    .windowAll(TumblingProcessingTimeWindows.of(Time.minutes(1)))
    .process(new ProcessAllWindowFunction<Tuple2<String, Long>, String, TimeWindow>() {
        public void process(Context ctx, 
                           Iterable<Tuple2<String, Long>> elements,
                           Collector<String> out) {
            // 排序并输出Top 10
            List<Tuple2<String, Long>> list = new ArrayList<>();
            elements.forEach(list::add);
            
            list.sort(Comparator.comparing((Tuple2<String, Long> t) -> t.f1).reversed());
            
            list.stream()
                .limit(10)
                .forEach(product -> 
                    out.collect("Product: " + product.f0 + ", Clicks: " + product.f1)
                );
        }
    });

场景3：实时告警（异常检测）

// 最近1分钟内错误率超过10%则告警，每10秒检查一次
logs
    .keyBy(log -> log.serviceId)
    .window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)))
    .process(new ProcessWindowFunction<Log, Alert, String, TimeWindow>() {
        public void process(
            String serviceId,
            Context ctx,
            Iterable<Log> logs,
            Collector<Alert> out
        ) {
            int totalCount = 0;
            int errorCount = 0;
            
            for (Log log : logs) {
                totalCount++;
                if (log.level.equals("ERROR")) {
                    errorCount++;
                }
            }
            
            double errorRate = totalCount == 0 ? 0 : (double) errorCount / totalCount;
            
            if (errorRate > 0.1) {  // 错误率超过10%
                out.collect(new Alert(
                    serviceId,
                    errorRate,
                    "High error rate: " + String.format("%.2f%%", errorRate * 100)
                ));
            }
        }
    });

场景4：流量监控（QPS统计）

// 统计最近10秒的QPS，每秒更新一次
requests
    .keyBy(req -> req.apiPath)
    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(1)))
    .aggregate(
        new AggregateFunction<Request, Long, Double>() {
            public Long createAccumulator() { return 0L; }
            public Long add(Request req, Long count) { return count + 1; }
            public Double getResult(Long count) { return count / 10.0; }  // QPS
            public Long merge(Long a, Long b) { return a + b; }
        }
    )
    .map(qps -> "QPS: " + String.format("%.2f", qps))
    .print();

/* 效果：
每秒输出最近10秒的平均QPS
更平滑的QPS曲线
*/

滑动步长的选择策略

不同步长的效果对比

// 窗口大小固定为60秒

// 步长60秒（滚动窗口）
.window(SlidingEventTimeWindows.of(Time.seconds(60), Time.seconds(60)))
// 特点：无重叠，每分钟统计一次

// 步长30秒（50%重叠）
.window(SlidingEventTimeWindows.of(Time.seconds(60), Time.seconds(30)))
// 特点：每30秒更新，数据属于2个窗口

// 步长10秒（高频更新）
.window(SlidingEventTimeWindows.of(Time.seconds(60), Time.seconds(10)))
// 特点：每10秒更新，数据属于6个窗口

// 步长1秒（极高频）
.window(SlidingEventTimeWindows.of(Time.seconds(60), Time.seconds(1)))
// 特点：每秒更新，数据属于60个窗口（计算开销大）

选择建议

步长选择更新频率计算开销适用场景
大步长（接近窗口大小）低低离线分析
中等步长（窗口大小的1/2）中中实时报表
小步长（窗口大小的1/10）高高实时监控
极小步长（秒级）极高极高高精度告警

性能优化考虑

1. 数据重复计算

// 滑动窗口的计算开销

// 滚动窗口（无重叠）
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
// 每条数据计算1次

// 滑动窗口（50%重叠）
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
// 每条数据计算2次（属于2个窗口）

// 滑动窗口（90%重叠）
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(1)))
// 每条数据计算10次（属于10个窗口）

// ⚠️ 计算开销 = 窗口大小 / 滑动步长

2. 状态存储开销

// 使用增量聚合减少状态存储

// ❌ 全量窗口函数：需要缓存所有数据
.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.seconds(10)))
.process(new ProcessWindowFunction<...>() {
    // 需要缓存10分钟的数据
    // 状态大小：O(窗口内元素数量 × 窗口数量)
});

// ✅ 增量聚合：只保存累积状态
.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.seconds(10)))
.reduce((v1, v2) -> v1 + v2);
// 状态大小：O(窗口数量) - 每个窗口只保存一个累积值

3. 增量聚合 + 全量窗口函数结合

// 最佳实践：既高效又灵活
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.aggregate(
    // 增量聚合（内存高效）
    new AggregateFunction<Event, Long, Long>() {
        public Long createAccumulator() { return 0L; }
        public Long add(Event event, Long acc) { return acc + 1; }
        public Long getResult(Long acc) { return acc; }
        public Long merge(Long a, Long b) { return a + b; }
    },
    // 全量窗口函数（获取窗口信息）
    new ProcessWindowFunction<Long, String, String, TimeWindow>() {
        public void process(String key, Context ctx, 
                           Iterable<Long> counts, Collector<String> out) {
            Long count = counts.iterator().next();
            out.collect(String.format(
                "Key: %s, Window: [%d-%d), Count: %d",
                key, ctx.window().getStart(), ctx.window().getEnd(), count
            ));
        }
    }
);

完整对比总结

窗口类型窗口大小滑动步长重叠度数据计算次数适用场景
滚动窗口10s10s0%1次独立统计
滑动窗口（50%重叠）10s5s50%2次移动平均
滑动窗口（80%重叠）10s2s80%5次实时监控
滑动窗口（90%重叠）10s1s90%10次高频告警

关键要点总结

  1. ✅ 滑动窗口特点：窗口可重叠，数据可属于多个窗口
  2. ✅ 两个参数：窗口大小（size）和滑动步长（slide）
  3. ✅ 重叠计算：slide < size时窗口重叠，数据被重复计算
  4. ✅ 事件时间vs处理时间：与滚动窗口相同的区别
  5. ✅ 偏移量：用于时区调整，实现跨时区统计
  6. ⚠️ 性能开销：计算次数 = size / slide
  7. ⚠️ 使用场景：移动平均、实时监控、趋势分析
  8. ⚠️ 优化建议：使用增量聚合函数减少状态存储
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区