再来一遍 Flink Ton案例代码

超、凢脫俗 2023-06-21 03:52 2阅读 0赞

一，windowAll  --它是一个task，请详细看官网对这个算子的说明

如果数据量太大的话，可以使用2个window优化

> `DataStream<Tuple2<String, Integer>> ret = wcount                 .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(20)))                                   .process(new TopNAllFunction(5));//计算该窗口TopN`

> private static class TopNAllFunction  
>             extends  
>             ProcessAllWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> \{
> 
>         private int topSize = 10;
> 
>         public TopNAllFunction(int topSize) \{  
>             // TODO Auto-generated constructor stub
> 
>             this.topSize = topSize;  
>         \}
> 
>         @Override  
>         public void process(  
>                 ProcessAllWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow>.Context arg0,  
>                 Iterable<Tuple2<String, Integer>> input,  
>                 Collector<Tuple2<String, Integer>> out) throws Exception \{  
>             // TODO Auto-generated method stub
> 
>             TreeMap<Integer, Tuple2<String, Integer>> treemap = new TreeMap<Integer, Tuple2<String, Integer>>(  
>                     new Comparator<Integer>() \{
> 
>                         @Override  
>                         public int compare(Integer y, Integer x) \{  
>                             // TODO Auto-generated method stub  
>                             return (x < y) ? -1 : 1;  
>                         \}
> 
>                     \}); //treemap按照key降序排列，相同count值不覆盖
> 
>             for (Tuple2<String, Integer> element : input) \{  
>                 treemap.put(element.f1, element);  
>                 if (treemap.size() > topSize) \{ //只保留前面TopN个元素  
>                     treemap.pollLastEntry();  
>                 \}  
>             \}
> 
>             for (Entry<Integer, Tuple2<String, Integer>> entry : treemap  
>                     .entrySet()) \{  
>                 out.collect(entry.getValue());  
>             \}
> 
>         \}
> 
>     \}

二，正常使用window算子

>   
> dstream.keyBy(new TupleKeySelectorByStart()) // 按照首字母分组  
>                 .window(TumblingProcessingTimeWindows.of(Time.seconds(20))) //20s窗口统计上游数据  
>                 .process(new TopNFunction(5)) //分组TopN统计  
>     private static class TupleKeySelectorByStart implements  
>             KeySelector<Tuple2<String, Integer>, String> \{
> 
>         @Override  
>         public String getKey(Tuple2<String, Integer> value) throws Exception \{  
>             // TODO Auto-generated method stub  
>             return value.f0.substring(0, 1); //取首字母做key  
>         \}
> 
>     \}  
> /\*\*  
>      \*   
>      \*针对keyby window的TopN函数，继承自ProcessWindowFunction  
>      \*  
>      \*/  
>     private static class TopNFunction  
>             extends  
>             ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow> \{
> 
>         private int topSize = 10;
> 
>         public TopNFunction(int topSize) \{  
>             // TODO Auto-generated constructor stub  
>             this.topSize = topSize;  
>         \}
> 
>         @Override  
>         public void process(  
>                 String arg0,  
>                 ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>.Context arg1,  
>                 Iterable<Tuple2<String, Integer>> input,  
>                 Collector<Tuple2<String, Integer>> out) throws Exception \{  
>             // TODO Auto-generated method stub
> 
>             TreeMap<Integer, Tuple2<String, Integer>> treemap = new TreeMap<Integer, Tuple2<String, Integer>>(  
>                     new Comparator<Integer>() \{
> 
>                         @Override  
>                         public int compare(Integer y, Integer x) \{  
>                             // TODO Auto-generated method stub  
>                             return (x < y) ? -1 : 1;  
>                         \}
> 
>                     \});
> 
>             for (Tuple2<String, Integer> element : input) \{  
>                 treemap.put(element.f1, element);  
>                 if (treemap.size() > topSize) \{  
>                     treemap.pollLastEntry();  
>                 \}  
>             \}
> 
>             for (Entry<Integer, Tuple2<String, Integer>> entry : treemap  
>                     .entrySet()) \{  
>                 out.collect(entry.getValue());  
>             \}
> 
>         \}
> 
>     \}

三，map算子与flatMap的区别：

在很多正常开发业务里面，如果业务报错，或者有部分数据会被过滤，就是有flatMap，它等同于 map+filter，只下发你需要的数据。