巨人肩膀
改版通知

巨人肩膀网站已全新改版。若您仍依赖旧站功能或数据，欢迎联系我们，我们会协助处理。联系我们

← 返回专栏列表

Flink窗口与状态编程开发一

ckckck2025年1月10日9 浏览

Transformation

Reduce

reduce 对按 key 分类的数据流进行“滚动”压缩。可以将当前元素与前一个元素进行整合处理，并返回一个新值。可用于前后数据的计算或拼接。

例如下面例子，通过 id 字段进行分类，然后使用 reduce 进行压缩处理，每次将学生的名字字段进行拼接，年龄进行相加，返回一个新的对象：

java 复制代码 
SingleOutputStreamOperator operator = source  
    .map((MapFunction<String, Student>) value -> parseTokens2Object(parseString2Tokens(value)))  
    .keyBy((KeySelector<Student, Integer>) value -> value == null ? 0 : value.getId())  
    .reduce((ReduceFunction<Student>) (value1, value2) -> {  
        Student student = new Student();  
        student.setId(value1.getId() + value2.getId());  
        student.setName(value1.getName() + " || " + value2.getName());  
        student.setAge(value1.getAge() + value2.getAge());  
        return student;  
    });

输出结果：

复制代码 
16> Student(id=2, name=name2, age=22)  
11> Student(id=1, name=name1, age=21)  
15> Student(id=3, name=name13, age=28)  
16> Student(id=5, name=name10, age=25)  
16> Student(id=10, name=name10 || name15, age=55)  
16> Student(id=4, name=name2 || name7, age=44)  
16> Student(id=6, name=name2 || name7 || name12, age=71)  
...  
16> Student(id=15, name=name10 || name15 || name5, age=80)  
1> Student(id=12, name=name4 || name9 || name14, age=77)  
11> Student(id=3, name=name1 || name6 || name11, age=68)

从结果可以看到，id 相同的都分到同一个分区（测试中可以简单通过前面的线程 id 确认，属于同一分区处理），然后传入新对象，按照 reduce 的操作进行了处理，返回了拼装之后的对象。

Fold

滚动折叠。合并当前元素和上一个被折叠的值，输入值可以与返回值类型不一样。

例如数据流是一组数字 [1, 5, 7]，想要输出一个拼接后的字符串，可以通过下面进行处理：

java 复制代码 
// 标准格式  
keyedStream.fold(${initialValue}, (s1, s2) -> s1 + " || " + s2);  

SingleOutputStreamOperator operator = source  
    .map((MapFunction<String, Student>) value -> parseTokens2Object(parseString2Tokens(value)))  
    .keyBy("id")  
    .fold("start", new FoldFunction<Student, String>() {  
        @Override  
        public String fold(String accumulator, Student value) throws Exception {  
            return accumulator + " || " + value;  
        }  
    });

输出结果：

复制代码 
16> start || Student(id=2, name=name2, age=22)  
11> start || Student(id=1, name=name1, age=21)  
15> start || Student(id=3, name=name13, age=28)  
16> start || Student(id=2, name=name2, age=22) || Student(id=2, name=name12, age=27)  
1> start || Student(id=4, name=name9, age=24)  
11> start || Student(id=1, name=name1, age=21) || Student(id=1, name=name6, age=21)  
1> start || Student(id=4, name=name9, age=24) || Student(id=4, name=name14, age=29)  
16> start || Student(id=2, name=name2, age=22) || Student(id=2, name=name12, age=27) || Student(id=2, name=name7, age=22)  
...

从输出结果可以看到，初始值和每次处理的对象进行了拼接，最后返回的是折叠后的对象，不过该方法被标注为 @Deprecated，不建议继续使用。

Project

选择部分字段。注意，只对元组 Tuple 类型的输入流有效，输出的也是选择下标的新元组数据流。例如下面，选择下标 1 和 3 的属性：

java 复制代码 
DataStreamSource<Tuple4<String, Integer, Long, BigDecimal>> customSource = env.fromCollection(  
    Lists.newArrayList(  
        Tuple4.of("one", 1, 1L, BigDecimal.ONE),  
        Tuple4.of("two", 2, 2L, BigDecimal.ZERO),  
        Tuple4.of("three", 3, 3L, BigDecimal.TEN),  
        Tuple4.of("four", 4, 4L, BigDecimal.TEN)  
    )  
);  
// 分离下标 1，3 到新的数据流  
DataStream<Tuple2<Integer, BigDecimal>> tuple2DataStreamSource = customSource.project(1, 3);

输出结果：

复制代码 
[1, 1]  
[2, 0]  
[3, 10]  
[4, 10]

Window

窗口。该函数允许在已分区的 KeyedStream 上定义窗口。例如最近 5 秒内到达的数据：

java 复制代码 
dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)));  
dataStream.keyBy(0).window(SlidingEventTimeWindows.of(Time.seconds(3), Time.seconds(2)));

AllWindowedStream

在 DataStream 上定义窗口。与上面不同的是，此次定义的范围是所有流，windows 会根据某些特征（例如，最近 5 秒内到达的数据）对所有流事件进行划分窗口。

java 复制代码 
SingleOutputStreamOperator<Student> operator = source  
    .map((MapFunction<String, Student>) value -> parseTokens2Object(parseString2Tokens(value)))  
    .assignTimestampsAndWatermarks(new MyTimestampExtractor())  
    .windowAll(TumblingEventTimeWindows.of(Time.milliseconds(1)))  
    .apply(new AllWindowFunction<Student, Student, TimeWindow>(){...});
Window

Reduce、Apply、Process、Aggregate、AggregateFunction、ProcessWindowFunction

  • 全量函数：窗口先缓存所有元素，等到触发条件后对窗口内的全量元素执行计算。
  • 增量函数：窗口保存一份中间数据，每流入一个新元素，新元素与中间数据两两合一，生成新的中间数据。

增量聚合函数计算性能较高，占用存储空间少，主要因为基于中间状态的计算结果，窗口中只维护中间结果状态值，不需要缓存原始数据。而全量窗口函数使用的代价相对较高，性能比较弱，主要因为此时算子需要对所有属于该窗口的接入数据进行缓存，然后等到窗口触发的时候，对所有的原始数据进行汇总计算。

Function

apply 和 process 都是处理全量计算，但工作中正常用 process。process 更加底层，更加强大，有 open/close 生命周期方法，又可获取 RuntimeContext。可以自己定时触发计算的定时器，形成自己的定时任务在 processElement 方法定义定时器 context.timerService().registerEventTimeTimer(timestamp);，当定时器时间到达，会回调 onTimer() 方法的计算任务。Apply 只能作用窗口流，而 Process 既可以作用于窗口流也可以作用于普通流。

Process
  • reduce 接受两个相同类型的输入，生成一个同类型输出，所以泛型就一个 <T>。
  • maxBy、minBy、sum 这 3 个底层都是由 reduce 实现的。
  • aggregate 的输入值、中间结果值、输出值它们 3 个类型可以各不相同，泛型有 <T, ACC, R>。

归约函数（ReduceFunction）
将窗口中收集到的数据两两进行归约。当我们进行流处理时，就是要保存一个状态；每来一个新的数据，就和之前的聚合状态做归约，这样就实现了增量式的聚合。所有聚合的操作保存在 Flink 的状态内存中，因为它需要跨越多条记录，需要根据 key 保存状态。数据流入的过程，就是不断计算并更新 Flink 中保存的状态的过程。

ReduceFunction 可以解决大多数归约聚合的问题，但是这个接口有一个限制，就是聚合状态的类型、输出结果的类型都必须和输入数据类型一样。这就迫使我们必须在聚合前，先将数据转换（map）成预期结果类型；而在有些情况下，还需要对状态进行进一步处理才能得到输出结果，这时它们的类型可能不同，使用 ReduceFunction 就会非常麻烦。

使用 ReduceFunction 能够快速对两个相同类型的数据元素按照指定的方法进行聚合逻辑，实现 sum 功能。

聚合函数（AggregateFunction）
aggregate 需要指定一个 AggregateFunction 函数，可以看做 reduce 函数的通用版本，这里有三种类型 IN, ACC, OUT，分别代表输入类型，累加器类型，输出类型。

接口中有四个方法：

  • createAccumulator()：创建一个累加器，这就是为聚合创建一个初始状态，每个聚合任务只会调用一次。
  • add()：将输入的元素添加到累加器中，这就是聚合状态，对于新来的数据进行进一步聚合的过程。传入两个参数，当前新到来的数据 value，和当前的累加器 accumulator；返回一个新的累加器值，对聚合状态进行更新。
  • getResult()：从累加器提取聚合的输出结果。也就是说我们可以定义多个状态，然后基于这些聚合的状态计算出一个结果进行输出。比如计算平均，我们可以设置 sum 和 count 两个状态，最终调用这个方法时相除得到最终的结果。这个方法只在窗口要输出结果时调用。
  • merge()：合并两个累加器，并将合并后的状态作为一个累加器返回。这个方法只在需要合并窗口的场景下才会被调用；最常见的合并窗口的场景就是会话窗口。

与 reduce 相比，aggregate 的输入格式与输出格式可以不同。更加灵活。

ReduceFunction 统计每一小时用户的访问量

java 复制代码 
package com.rosh.flink.pojo;  
import lombok.AllArgsConstructor;  
import lombok.Data;  
import lombok.NoArgsConstructor;  

@AllArgsConstructor  
@NoArgsConstructor  
@Data  
public class UserPojo {  
    private Integer userId;  
    private String name;  
    private String uri;  
    private Long timestamp;  
}  

package com.rosh.flink.wartermark;  
import com.rosh.flink.pojo.UserPojo;  
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;  
import org.apache.flink.api.common.eventtime.WatermarkStrategy;  
import org.apache.flink.api.common.functions.MapFunction;  
import org.apache.flink.api.common.functions.ReduceFunction;  
import org.apache.flink.api.java.tuple.Tuple2;  
import org.apache.flink.streaming.api.datastream.DataStreamSource;  
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;  
import org.apache.flink.streaming.api.windowing.time.Time;  
import java.security.NoSuchAlgorithmException;  
import java.security.SecureRandom;  
import java.util.ArrayList;  
import java.util.List;  
import java.util.Random;  

public class WindowTS {  
    public static void main(String[] args) throws Exception {  
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
        env.setParallelism(1);  
        DataStreamSource<UserPojo> dataDS = env.fromCollection(getUserLists());  
        // 生成有序水位线  
        SingleOutputStreamOperator<UserPojo> orderStreamDS = dataDS.assignTimestampsAndWatermarks(WatermarkStrategy.<UserPojo>forMonotonousTimestamps()  
                .withTimestampAssigner(new SerializableTimestampAssigner<UserPojo>() {  
                    @Override  
                    public long extractTimestamp(UserPojo element, long recordTimestamp) {  
                        return element.getTimestamp();  
                    }  
                }));  
        // 聚合  
        SingleOutputStreamOperator<Tuple2<Integer, Long>> userDS = orderStreamDS.map(new MapFunction<UserPojo, Tuple2<Integer, Long>>() {  
            @Override  
            public Tuple2<Integer, Long> map(UserPojo value) throws Exception {  
                return Tuple2.of(value.getUserId(), 1L);  
            }  
        });  
        // 开窗统计每1小时用户访问了多少次  
        SingleOutputStreamOperator<Tuple2<Integer, Long>> resultDS = userDS.keyBy(tuple -> tuple.f0)  
                .window(TumblingEventTimeWindows.of(Time.hours(1)))  
                .reduce(new ReduceFunction<Tuple2<Integer, Long>>() {  
                    @Override  
                    public Tuple2<Integer, Long> reduce(Tuple2<Integer, Long> value1, Tuple2<Integer, Long> value2) throws Exception {  
                        value1.f1 = value1.f1 + value2.f1;  
                        return value1;  
                    }  
                });  
        resultDS.print();  
        env.execute("WarterMarkTest");  
    }  
    private static List<UserPojo> getUserLists() throws NoSuchAlgorithmException {  
        List<UserPojo> lists = new ArrayList<>();  
        Random random = SecureRandom.getInstanceStrong();  
        for (int i = 1; i <= 1000; i++) {  
            String uri = "/goods/" + i;  
            int userId = random.nextInt(10);  
            // 有序时间  
            UserPojo userPojo = new UserPojo(userId, "name" + userId, uri, (long) (1000 * i));  
            // 无序时间  
            lists.add(userPojo);  
        }  
        return lists;  
    }  
}

AggregateFunction 统计人均访问次数

java 复制代码 
package com.rosh.flink.wartermark;  
import com.rosh.flink.pojo.UserPojo;  
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;  
import org.apache.flink.api.common.eventtime.WatermarkStrategy;  
import org.apache.flink.api.common.functions.AggregateFunction;  
import org.apache.flink.api.java.tuple.Tuple2;  
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;  
import org.apache.flink.streaming.api.windowing.time.Time;  
import java.security.NoSuchAlgorithmException;  
import java.security.SecureRandom;  
import java.util.*;  

public class AggWindowTest {  
    public static void main(String[] args) throws Exception {  
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
        env.setParallelism(1);  
        SingleOutputStreamOperator<UserPojo> userDS = env.fromCollection(getUserLists()).assignTimestampsAndWatermarks(  
                WatermarkStrategy.<UserPojo>forMonotonousTimestamps()  
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserPojo>() {  
                            @Override  
                            public long extractTimestamp(UserPojo element, long recordTimestamp) {  
                                return element.getTimestamp();  
                            }  
                        })  
        );  
        // 统计5秒内，人均访问次数  
        SingleOutputStreamOperator<Double> resultDS = userDS.keyBy(key -> true)  
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))  
                .aggregate(new PeopleHourAvgCount());  
        resultDS.print("人均访问次数为：");  
        env.execute("AggWindowTest");  
    }  
    private static class PeopleHourAvgCount implements AggregateFunction<UserPojo, Tuple2<HashSet<Integer>, Long>, Double> {  
        @Override  
        public Tuple2<HashSet<Integer>, Long> createAccumulator() {  
            return Tuple2.of(new HashSet<>(), 0L);  
        }  
        @Override  
        public Tuple2<HashSet<Integer>, Long> add(UserPojo value, Tuple2<HashSet<Integer>, Long> accumulator) {  
            // distinct userId  
            accumulator.f0.add(value.getUserId());  
            // 次数+1  
            accumulator.f1 = accumulator.f1 + 1;  
            // 返回累加器  
            return accumulator;  
        }  
        @Override  
        public Double getResult(Tuple2<HashSet<Integer>, Long> accumulator) {  
            return accumulator.f1 * 1.0 / accumulator.f0.size();  
        }  
        @Override  
        public Tuple2<HashSet<Integer>, Long> merge(Tuple2<HashSet<Integer>, Long> a, Tuple2<HashSet<Integer>, Long> b) {  
            return null;  
        }  
    }  
    private static List<UserPojo> getUserLists() throws NoSuchAlgorithmException {  
        List<UserPojo> lists = new ArrayList<>();  
        Random random = SecureRandom.getInstanceStrong();  
        // 获取随机人数  
        int peopleCount = random.nextInt(20);  
        System.out.println("随机人数为:" + peopleCount);  
        for (int i = 1; i <= 1000; i++) {  
            String uri = "/goods/" + i;  
            int userId = random.nextInt(peopleCount);  
            // 有序时间  
            UserPojo userPojo = new UserPojo(userId, "name" + userId, uri, new Date().getTime());  
            // 无序时间  
            lists.add(userPojo);  
        }  
        return lists;  
    }  
}

全窗口函数

窗口操作中的另一大类就是全窗口函数。与增量聚合函数不同，全窗口函数需要先收集窗口中的数据，并在内部缓存起来，等到窗口要输出结果的时候再取出数据进行计算。这种计算方式相比于流处理是低效的，但是有的时候必须获取到所有数据才能计算，或者需要获取窗口的起始时间等，那么就必须使用全窗口函数。这是典型的批处理思想。

  1. 窗口函数（WindowFunction）
    WindowFunction 字面上就是“窗口函数”，他其实就是老版本的通用窗口函数接口。我们可以基于 WindowedStream 调用 apply 方法，传入一个 WindowFunction 的实现类。WindowFunction 可用的功能较少，一般使用 ProcessWindowFunction。

ProcessWindowFunction 是 Window API 中最底层的通用窗口函数接口。之所以说它“最底层”，是因为除了可以拿到窗口中的所有数据之外，ProcessWindowFunction 还可以获取到一个“上下文对象”（Context）。这个上下文对象非常强大，不仅能够获取窗口信息，还可以访问当前的时间和状态信息。这里的时间就包括了处理时间（processing time）和事件时间水位线（event time watermark）。这就使得 ProcessWindowFunction 更加灵活、功能更加丰富。事实上，ProcessWindowFunction 是 Flink 底层 API——处理函数（process function）中的一员。

统计10秒访问UV

java 复制代码 
package com.rosh.flink.wartermark;  
import com.rosh.flink.pojo.UserPojo;  
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;  
import org.apache.flink.api.common.eventtime.WatermarkStrategy;  
import org.apache.flink.streaming.api.datastream.DataStreamSource;  
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;  
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;  
import org.apache.flink.streaming.api.windowing.time.Time;  
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;  
import org.apache.flink.util.Collector;  
import java.security.NoSuchAlgorithmException;  
import java.security.SecureRandom;  
import java.sql.Timestamp;  
import java.util.*;  

public class ProcessWindowTest {  
    public static void main(String[] args) throws Exception {  
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
        env.setParallelism(1);  
        DataStreamSource<UserPojo> userDS = env.fromCollection(getUserLists());  
        // 水位线  
        SingleOutputStreamOperator<UserPojo> watermarks = userDS.assignTimestampsAndWatermarks(WatermarkStrategy.<UserPojo>forMonotonousTimestamps()  
                .withTimestampAssigner(new SerializableTimestampAssigner<UserPojo>() {  
                    @Override  
                    public long extractTimestamp(UserPojo element, long recordTimestamp) {  
                        return element.getTimestamp();  
                    }  
                }));  
        // 开窗10秒UV统计  
        SingleOutputStreamOperator<String> resultDS = watermarks.keyBy(key -> true)  
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))  
                .process(new UserUVCount());  
        resultDS.print("UV:");  
        env.execute("ProcessWindowTest");  
    }  
    private static class UserUVCount extends ProcessWindowFunction<UserPojo, String, Boolean, TimeWindow> {  
        @Override  
        public void process(Boolean aBoolean, ProcessWindowFunction<UserPojo, String, Boolean, TimeWindow>.Context context, Iterable<UserPojo> elements, Collector<String> out) throws Exception {  
            // 用户集合  
            HashSet<Integer> hashSet = new HashSet<>();  
            for (UserPojo user : elements) {  
                hashSet.add(user.getUserId());  
            }  
            // 获取时间信息  
            long start = context.window().getStart();  
            long end = context.window().getEnd();  
            String rs = "窗口信息，startTime：" + new Timestamp(start) + "，endTime: " + new Timestamp(end) + "，用户访问的次数为：" + hashSet.size();  
            out.collect(rs);  
        }  
    }  
    private static List<UserPojo> getUserLists() throws NoSuchAlgorithmException {  
        List<UserPojo> lists = new ArrayList<>();  
        Random random = SecureRandom.getInstanceStrong();  
        int userCount = random.nextInt(100);  
        for (int i = 1; i <= 1000; i++) {  
            String uri = "/goods/" + i;  
            int userId = random.nextInt(userCount);  
            // 有序时间  
            UserPojo userPojo = new UserPojo(userId, "name" + userId, uri, new Date().getTime());  
            // 无序时间  
            lists.add(userPojo);  
        }  
        return lists;  
    }  
}

增量聚合和全窗口函数的结合使用

增量聚合函数处理计算会更高效。全窗口函数的优势在于提供了更多的信息，可以认为是更加“通用”的窗口操作。所以在实际应用中，我们往往希望兼具这两者的优点，把它们结合在一起使用。Flink 的 Window API 就给我们实现了这样的用法。

在调用 WindowedStream 的 reduce() 和 aggregate() 方法时，只是简单地直接传入了一个 ReduceFunction 或 AggregateFunction 进行增量聚合。除此之外，其实还可以传入第二个参数：一个全窗口函数，可以是 WindowFunction 或者 ProcessWindowFunction。

java 复制代码 
// ReduceFunction 与 WindowFunction 结合  
public <R> SingleOutputStreamOperator<R> reduce(ReduceFunction<T> reduceFunction, WindowFunction<T, R, K, W> function)  

// ReduceFunction 与 ProcessWindowFunction 结合  
public <R> SingleOutputStreamOperator<R> reduce(ReduceFunction<T> reduceFunction, ProcessWindowFunction<T, R, K, W> function)  

// AggregateFunction 与 WindowFunction 结合  
public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T, ACC, V> aggFunction, WindowFunction<V, R, K, W> windowFunction)  

// AggregateFunction 与 ProcessWindowFunction 结合  
public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T, ACC, V> aggFunction, ProcessWindowFunction<V, R, K, W> windowFunction)

这样调用的处理机制是：基于第一个参数（增量聚合函数）来处理窗口数据，每来一个数据就做一次聚合；等到窗口需要触发计算时，则调用第二个参数（全窗口函数）的处理逻辑输出结果。需要注意的是，这里的全窗口函数就不再缓存所有数据了，而是直接将增量聚合函数的结果拿来当作了 Iterable 类型的输入。一般情况下，这时的可迭代集合中就只有一个元素了。

ProcessWindowFunction 和其他窗口函数可以一起组合使用，满足你的一切需求，因为将 ProcessWindowFunction 用于简单的聚合（比如 count）是非常低效的。ProcessWindowFunction 可以与 ReduceFunction、AggregateFunction 或 FoldFunction 组合，以便在元素到达窗口时就增量地聚合它们，减少 ProcessWindowFunction 处理的数据量。当窗口关闭时，ProcessWindowFunction 将提供聚合的结果。这允许它在访问 ProcessWindowFunction 的附加窗口元信息时以增量方式计算窗口。

统计10秒的url浏览量

java 复制代码 
package com.rosh.flink.wartermark;  
import com.alibaba.fastjson.JSONObject;  
import com.rosh.flink.pojo.UserPojo;  
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;  
import org.apache.flink.api.common.eventtime.WatermarkStrategy;  
import org.apache.flink.api.common.functions.MapFunction;  
import org.apache.flink.api.common.functions.ReduceFunction;  
import org.apache.flink.api.java.tuple.Tuple2;  
import org.apache.flink.streaming.api.datastream.DataStreamSource;  
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;  
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;  
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;  
import org.apache.flink.streaming.api.windowing.time.Time;  
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;  
import org.apache.flink.util.Collector;  
import java.security.NoSuchAlgorithmException;  
import java.security.SecureRandom;  
import java.sql.Timestamp;  
import java.util.ArrayList;  
import java.util.Date;  
import java.util.List;  
import java.util.Random;  

public class UrlWindowTest {  
    public static void main(String[] args) throws Exception {  
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
        env.setParallelism(1);  
        // 读取数据源  
        DataStreamSource<UserPojo> userDS = env.fromCollection(getUserLists());  
        // 水位线  
        SingleOutputStreamOperator<UserPojo> waterDS = userDS.assignTimestampsAndWatermarks(WatermarkStrategy.<UserPojo>forMonotonousTimestamps()  
                .withTimestampAssigner(new SerializableTimestampAssigner<UserPojo>() {  
                    @Override  
                    public long extractTimestamp(UserPojo element, long recordTimestamp) {  
                        return element.getTimestamp();  
                    }  
                }));  
        // url count  
        SingleOutputStreamOperator<Tuple2<String, Long>> urlDS = waterDS.map(new MapFunction<UserPojo, Tuple2<String, Long>>() {  
            @Override  
            public Tuple2<String, Long> map(UserPojo value) throws Exception {  
                return Tuple2.of(value.getUri(), 1L);  
            }  
        });  
        SingleOutputStreamOperator<JSONObject> resultDS = urlDS.keyBy(data -> data.f0)  
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))  
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {  
                    @Override  
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {  
                        value1.f1 = value1.f1 + value2.f1;  
                        return value1;  
                    }  
                }, new WindowFunction<Tuple2<String, Long>, JSONObject, String, TimeWindow>() {  
                    @Override  
                    public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<JSONObject> out) throws Exception {  
                        Tuple2<String, Long> tuple2 = input.iterator().next();  
                        JSONObject jsonObject = new JSONObject();  
                        jsonObject.put("url", tuple2.f0);  
                        jsonObject.put("count", tuple2.f1);  
                        new Timestamp(window.getStart());  
                        jsonObject.put("startTime", new Timestamp(window.getStart()).toString());  
                        jsonObject.put("endTime", new Timestamp(window.getEnd()).toString());  
                        out.collect(jsonObject);  
                    }  
                });  
        resultDS.print();  
        env.execute("UrlWindowTest");  
    }  
    private static List<UserPojo> getUserLists() throws NoSuchAlgorithmException {  
        List<UserPojo> lists = new ArrayList<>();  
        Random random = SecureRandom.getInstanceStrong();  
        for (int i = 1; i <= 1000; i++) {  
            // 随机生成userId、goodId  
            int userId = random.nextInt(100);  
            int goodId = random.nextInt(50);  
            String uri = "/goods/" + goodId;  
            // 有序时间  
            UserPojo userPojo = new UserPojo(userId, "name" + userId, uri, new Date().getTime());  
            // 无序时间  
            lists.add(userPojo);  
        }  
        return lists;  
    }  
}

end