0

Flink是下一代大数据计算平台,可处理流计算和批量计算。《Flink-1.9流计算开发:十一、count-window-Sliding》cosmozhu写的本系列文章的第十一篇。通过简单的DEMO来演示count-window-Sliding时间窗口函数执行的效果 。

需求

本篇文章,我们使用count-window-Sliding时间窗口,每5个订单统计一次最近20个订单的订单数量。

解决方案

public class StreamTest {
    private static final Logger LOG = LoggerFactory.getLogger(StreamTest.class);
    private static final String[] TYPE = { "苹果", "梨", "西瓜", "葡萄", "火龙果" };

    @SuppressWarnings("deprecation")
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
        DataStreamSource<Tuple2<String, Integer>> orderSource = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
            private volatile boolean isRunning = true;
            private final Random random = new Random();
            @Override
            public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                while (isRunning) {
                    TimeUnit.SECONDS.sleep(1);
                    ctx.collect(Tuple2.of(TYPE[random.nextInt(TYPE.length)], 1));
                }
            }
            @Override
            public void cancel() {
                isRunning = false;
            }

        }, "order-info");
        //这里只为将DataStream → KeyedStream,用空字符串做分区键。所有数据为相同分区
        orderSource
        //每5个订单统计一次最近20个订单的订单数量
        .countWindowAll(20,5)
        //这里用HashMap做暂存器
        .fold(new HashMap<String, Integer>(), new FoldFunction<Tuple2<String,Integer>, Map<String, Integer>>() {
            @Override
            public Map fold(Map<String, Integer> accumulator, Tuple2<String, Integer> value) throws Exception {
                accumulator.put(value.f0, (Integer)accumulator.getOrDefault(value.f0, 0)+value.f1);
                return accumulator;
            }
        })
        .print();

        env.execute("Flink Streaming Java API Skeleton");
    }
}

执行效果

4> {苹果=1, 梨=2, 西瓜=2}
1> {苹果=3, 葡萄=1, 火龙果=1, 梨=3, 西瓜=2}
2> {苹果=5, 葡萄=1, 火龙果=2, 梨=4, 西瓜=3}
3> {苹果=6, 葡萄=3, 火龙果=2, 梨=5, 西瓜=4}
4> {苹果=6, 葡萄=6, 火龙果=2, 梨=4, 西瓜=2}
1> {苹果=5, 葡萄=6, 火龙果=1, 梨=5, 西瓜=3}
2> {苹果=5, 葡萄=7, 梨=6, 西瓜=2}
3> {葡萄=6, 苹果=6, 梨=7, 西瓜=1}
4> {葡萄=4, 苹果=5, 火龙果=1, 梨=7, 西瓜=3}
1> {葡萄=6, 苹果=5, 火龙果=1, 梨=5, 西瓜=3}
2> {苹果=3, 葡萄=6, 火龙果=2, 梨=4, 西瓜=5}
3> {葡萄=5, 苹果=3, 火龙果=2, 梨=4, 西瓜=6}
4> {葡萄=4, 苹果=3, 火龙果=3, 梨=4, 西瓜=6}

小结

这里解释一下count-Sliding这个时间窗口,如图所示:
countsliding

代码地址

https://github.com/chaoxxx/learn-flink-stream-api/blob/master/src/main/java/fun/cosmozhu/session11/StreamTest.java

作者:cosmozhu --90后的老父亲,专注于保护地球的程序员
个人网站:https://www.cosmozhu.fun
欢迎转载,转载时请注明出处。