Flink aggregate()函数使用遇到的一点小问题

深碍√TFBOYSˉ_ 2024-04-18 17:23 61阅读 0赞

在使用aggregate()函数的时候，一直报错，先贴代码吧

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

一直不知道是什么原因，只是说方法不匹配，想想可能是keyBy函数的返回key值类型不对，改成String类型，果然是这个问题。

因为是java写，所以要注意这个问题：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

scala好像不需要注意这个点 ：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

使用对象：

>     package com.aliyun.market;
>     
>     
>     import com.alibaba.fastjson.JSON;
>     import org.apache.flink.api.common.functions.AggregateFunction;
>     import org.apache.flink.api.common.functions.FilterFunction;
>     import org.apache.flink.api.common.functions.MapFunction;
>     import org.apache.flink.api.common.serialization.SimpleStringSchema;
>     import org.apache.flink.api.common.state.ListState;
>     import org.apache.flink.api.common.state.ListStateDescriptor;
>     import org.apache.flink.api.java.tuple.Tuple;
>     import org.apache.flink.api.java.tuple.Tuple1;
>     import org.apache.flink.api.java.utils.ParameterTool;
>     import org.apache.flink.configuration.Configuration;
>     import org.apache.flink.streaming.api.TimeCharacteristic;
>     import org.apache.flink.streaming.api.datastream.DataStream;
>     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
>     import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
>     import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
>     import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
>     import org.apache.flink.streaming.api.windowing.time.Time;
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
>     import org.apache.flink.util.Collector;
>     
>     import java.sql.Timestamp;
>     import java.util.ArrayList;
>     import java.util.Comparator;
>     import java.util.List;
>     import java.util.Properties;
>     
>     
>     /**
>      * 热门商品topN统计
>      * 数据下载 curl https://raw.githubusercontent.com/wuchong/my-flink-project/master/src/main/resources/UserBehavior.csv > UserBehavior.csv
>      * 数据在resource 下 UserBehavior.csv
>      * 参考： http://wuchong.me/blog/2018/11/07/use-flink-calculate-hot-items/
>      * <p>
>      * 知识点： aggregate函数 是专门来做统计的，一般跟着keyBy（）后面
>      * <p>
>      * 官网使用 ： aggregate(SUM, 0).and(MIN, 2)
>      */
>     public class HotTopDemo {
>         public static void main(String[] args) throws Exception {
>     
>             // todo 1，读取kafka数据
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>     
>             //todo 获取kafka的配置属性
>             args = new String[]{"--input-topic", "user_behavior", "--bootstrap.servers", "node2.hadoop:9091,node3.hadoop:9091",
>                     "--zookeeper.connect", "node1.hadoop:2181,node2.hadoop:2181,node3.hadoop:2181", "--group.id", "cc2"};
>     
>             ParameterTool parameterTool = ParameterTool.fromArgs(args);
>             env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>     
>     
>     //
>             Properties pros = parameterTool.getProperties();
>     //        //todo 指定输入数据为kafka topic
>             DataStream<String> kafkaDstream = env.addSource(new FlinkKafkaConsumer010<String>(
>                     pros.getProperty("input-topic"),
>                     new SimpleStringSchema(),
>                     pros).setStartFromLatest()
>     
>             ).setParallelism(1);
>     
>     
>     
>             //todo 我们用 PojoCsvInputFormat 创建输入源。
>             DataStream<UserBehavior> dataDstream =kafkaDstream.map(new MapFunction<String, UserBehavior>() {
>                 @Override
>                 public UserBehavior map(String input) throws Exception {
>                     return  JSON.parseObject(input, UserBehavior.class);
>                 }
>             });
>     
>             //给数据加上了一个水印
>             DataStream<UserBehavior> timedData = dataDstream
>                     .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
>                         @Override
>                         public long extractAscendingTimestamp(UserBehavior userBehavior) {
>                             // 原始数据单位秒 ，将其转成毫秒
>                             return userBehavior.category_id * 1000;
>                         }
>                     });
>     
>             //过滤出点击事件
>     
>             DataStream<UserBehavior> pvData = timedData.filter(new FilterFunction<UserBehavior>() {
>                 @Override
>                 public boolean filter(UserBehavior userBehavior) throws Exception {
>                     return userBehavior.behavior.equals("pv");
>                 }
>             });
>     
>             //todo 窗口统计点击量,设置窗口大小为1个小时，5分钟滑动一次
>             //由于要每隔5分钟统计一次最近一小时每个商品的点击量，所以窗口大小是一小时，每隔5分钟滑动一次。即分别要统计 [09:00, 10:00), [09:05, 10:05), [09:10, 10:10)… 等窗口的商品点击量。是一个常见的滑动窗口需求（Sliding Window）。
>     
>             DataStream<ItemViewCount> windowedData = pvData.keyBy("item_id") //按字段分组
>                     .timeWindow(Time.seconds(20), Time.seconds(10))
>                     .aggregate(new CountAgg(), new WindowResultFunction()); //使用aggregate做增量聚合统计
>     
>     //        TopN 计算最热门商品
>             DataStream<String> topItems = windowedData
>                     .keyBy("windowEnd")
>                     .process(new TopNHotItems(2));  // 求点击量前3名的商品
>             topItems.print();
>             env.execute("Hot Items Job");
>     
>         }
>     
>     
>     
>     
>     
>     
>     
>     
>     
>     
>     
>     
>     
>     
>     
>         //todo 功能是统计窗口中的条数，即遇到一条数据就加一
>     
>         public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {
>     
>             @Override
>             public Long createAccumulator() { //创建一个数据统计的容器，提供给后续操作使用。
>                 return 0L;
>             }
>     
>             @Override
>             public Long add(UserBehavior userBehavior, Long acc) { //每个元素被添加进窗口的时候调用。
>                 return acc + 1;
>             }
>     
>             @Override
>             public Long getResult(Long acc) {
>                 ;//窗口统计事件触发时调用来返回出统计的结果。
>                 return acc;
>             }
>     
>             @Override
>             public Long merge(Long acc1, Long acc2) { //只有在当窗口合并的时候调用,合并2个容器
>                 return acc1 + acc2;
>             }
>         }
>     
>         // todo 指定格式输出 ：将每个 key每个窗口聚合后的结果带上其他信息进行输出  进入的数据为Long 返回 ItemViewCount对象
>         public static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {
>     
>             @Override
>             public void apply(
>                     Tuple key,  // 窗口的主键，即 itemId
>                     TimeWindow window,  // 窗口
>                     Iterable<Long> aggregateResult, // 聚合函数的结果，即 count 值
>                     Collector<ItemViewCount> collector  // 输出类型为 ItemViewCount
>             ) throws Exception {
>                 Long itemId = ((Tuple1<Long>) key).f0;
>                 Long count = aggregateResult.iterator().next();
>                 collector.collect(ItemViewCount.of(itemId, window.getEnd(), count));
>             }
>         }
>     
>         /**
>          * 商品点击量(窗口操作的输出类型)
>          */
>         public static class ItemViewCount {
>             public long itemId;     // 商品ID
>             public long windowEnd;  // 窗口结束时间戳
>             public long viewCount;  // 商品的点击量
>     
>             public static ItemViewCount of(long itemId, long windowEnd, long viewCount) {
>                 ItemViewCount result = new ItemViewCount();
>                 result.itemId = itemId;
>                 result.windowEnd = windowEnd;
>                 result.viewCount = viewCount;
>                 return result;
>             }
>         }
>     
>         /** 求某个窗口中前 N 名的热门点击商品，key 为窗口时间戳，输出为 TopN 的结果字符串 */
>         public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String> {
>     
>             private final int topSize;
>     
>             public TopNHotItems(int topSize) {
>                 this.topSize = topSize;
>             }
>     
>             // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算
>             private ListState<ItemViewCount> itemState;
>     
>             @Override
>             public void open(Configuration parameters) throws Exception {
>                 super.open(parameters);
>                 // 状态的注册
>                 ListStateDescriptor<ItemViewCount> itemsStateDesc = new ListStateDescriptor<>(
>                         "itemState-state",
>                         ItemViewCount.class);
>                 itemState = getRuntimeContext().getListState(itemsStateDesc);
>             }
>     
>             @Override
>             public void processElement(
>                     ItemViewCount input,
>                     Context context,
>                     Collector<String> collector) throws Exception {
>     
>                 // 每条数据都保存到状态中
>                 itemState.add(input);
>                 // 注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收齐了属于windowEnd窗口的所有商品数据
>                 context.timerService().registerEventTimeTimer(input.windowEnd + 1);
>             }
>     
>             @Override
>             public void onTimer(
>                     long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
>                 // 获取收到的所有商品点击量
>                 List<ItemViewCount> allItems = new ArrayList<>();
>                 for (ItemViewCount item : itemState.get()) {
>                     allItems.add(item);
>                 }
>                 // 提前清除状态中的数据，释放空间
>                 itemState.clear();
>                 // 按照点击量从大到小排序
>                 allItems.sort(new Comparator<ItemViewCount>() {
>                     @Override
>                     public int compare(ItemViewCount o1, ItemViewCount o2) {
>                         return (int) (o2.viewCount - o1.viewCount);
>                     }
>                 });
>                 // 将排名信息格式化成 String, 便于打印
>                 StringBuilder result = new StringBuilder();
>                 result.append("====================================\n");
>                 result.append("时间: ").append(new Timestamp(timestamp-1)).append("\n");
>                 for (int i=0;i<topSize;i++) {
>                     ItemViewCount currentItem = allItems.get(i);
>                     // No1:  商品ID=12224  浏览量=2413
>                     result.append("No").append(i).append(":")
>                             .append("  商品ID=").append(currentItem.itemId)
>                             .append("  浏览量=").append(currentItem.viewCount)
>                             .append("\n");
>                 }
>                 result.append("====================================\n\n");
>     
>                 out.collect(result.toString());
>             }
>         }
>     }

使用fastJson

>     package com.aliyun.market;
>     
>     import com.alibaba.fastjson.JSONObject;
>     import org.apache.flink.api.common.functions.AggregateFunction;
>     import org.apache.flink.api.common.functions.FilterFunction;
>     import org.apache.flink.api.common.functions.MapFunction;
>     import org.apache.flink.api.common.serialization.SimpleStringSchema;
>     import org.apache.flink.api.java.functions.KeySelector;
>     import org.apache.flink.api.java.utils.ParameterTool;
>     import org.apache.flink.streaming.api.TimeCharacteristic;
>     import org.apache.flink.streaming.api.datastream.DataStream;
>     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
>     import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
>     import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
>     import org.apache.flink.streaming.api.windowing.time.Time;
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
>     import org.apache.flink.util.Collector;
>     
>     import java.util.Properties;
>     
>     /**
>      * 商场实时数据统计,基于Flink 1.9版本
>      */
>     public class MarketStreamCount {
>         public static void main(String[] args) {
>     
>     
>             // todo 1，读取kafka数据
>             final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(8);
>             //todo 获取kafka的配置属性
>             args = new String[]{"--input-topic", "user_behavior", "--bootstrap.servers", "node2.hadoop:9091,node3.hadoop:9091",
>                     "--zookeeper.connect", "node1.hadoop:2181,node2.hadoop:2181,node3.hadoop:2181", "--group.id", "cc2"};
>     
>             ParameterTool parameterTool = ParameterTool.fromArgs(args);
>             env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>     
>              Properties pros = parameterTool.getProperties();
>     //        //todo 指定输入数据为kafka topic
>             DataStream<String> kafkaDstream = env.addSource(new FlinkKafkaConsumer010<String>(
>                     pros.getProperty("input-topic"),
>                     new SimpleStringSchema(),
>                     pros).setStartFromEarliest()
>     
>             ).setParallelism(1);
>     
>             //todo 2,每层实时顾客人数，实时顾客总数，一天的实时顾客总数
>     
>             // 转成json
>             DataStream<JSONObject> kafkaDstream2 = kafkaDstream.map(new MapFunction<String, JSONObject>() {
>                 @Override
>                 public JSONObject map(String input) throws Exception {
>                     JSONObject inputJson = null;
>                     try {
>                         inputJson = JSONObject.parseObject(input);
>                         return inputJson;
>                     } catch (Exception e) {
>                         e.printStackTrace();
>     //                    return null;
>                     }
>                     return inputJson;
>                 }
>             }).filter(new FilterFunction<JSONObject>() {
>                 @Override
>                 public boolean filter(JSONObject jsonObject) throws Exception {
>                     if (jsonObject!=null){
>                         return true;
>                     }
>                     return false;
>                 }
>             });
>             //给数据加上了一个水印
>             DataStream<JSONObject> timedData = kafkaDstream2
>                     .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<JSONObject>() {
>                         @Override
>                         public long extractAscendingTimestamp(JSONObject json) {
>                             // 原始数据单位秒 ，将其转成毫秒
>                             return json.getLong("category_id") * 1000;
>                         }
>                     });
>     
>             //过滤出点击事件
>             // 实时客人数，各个层级
>             DataStream<JSONObject> windowedData = timedData.keyBy(new KeySelector<JSONObject, String>() {
>                 @Override
>                 public String getKey(JSONObject jsonObject) throws Exception {
>                     return  jsonObject.getString("user_id");
>                 }
>             }).timeWindow(Time.seconds(5L),Time.seconds(1L))
>                     //使用aggregate做增量聚合统计
>                     .aggregate(new CountAgg(),new WindowResultFunction());
>     
>     
>     
>             windowedData.print();
>             try {
>                 env.execute();
>             } catch (Exception e) {
>                 e.printStackTrace();
>             }
>         }
>     
>     
>         //todo 功能是统计窗口中的条数，即遇到一条数据就加一
>     
>         public static class CountAgg implements AggregateFunction<JSONObject, Long, Long> {
>     
>             @Override
>             public Long createAccumulator() { //创建一个数据统计的容器，提供给后续操作使用。
>                 return 0L;
>             }
>     
>             @Override
>             public Long add(JSONObject json, Long acc) { //每个元素被添加进窗口的时候调用。
>                 return acc + 1;
>             }
>     
>             @Override
>             public Long getResult(Long acc) {
>                 //窗口统计事件触发时调用来返回出统计的结果。
>                 return acc;
>             }
>     
>             @Override
>             public Long merge(Long acc1, Long acc2) { //只有在当窗口合并的时候调用,合并2个容器
>                 return acc1 + acc2;
>             }
>         }
>     
>         // todo 指定格式输出
>         public static class WindowResultFunction implements WindowFunction<Long, JSONObject, String, TimeWindow> {
>     
>             @Override
>             public void apply(
>                     String key,  // 窗口的主键，即 itemId
>                     TimeWindow window,  // 窗口
>                     Iterable<Long> aggregateResult, // 聚合函数的结果，即 count 值
>                     Collector<JSONObject> collector  // 输出类型为 ItemViewCount
>             ) throws Exception {
>     
>                 Long count = aggregateResult.iterator().next();
>                 //窗口结束时间
>                 long end = window.getEnd();
>                 JSONObject json = new JSONObject();
>                 json.put("key",key);
>                 json.put("count",count);
>                 json.put("end",end);
>                 collector.collect(json);
>             }
>         }
>     
>     }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/0814f82df3f7401ea3aa338822e580bb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/39bbeeec316a405990e25ce29ec733c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/a63b27506481442e9137ed6d1a2f3913.png