flink求topn

Flink可以使用ProcessFunction实现TopN操作。下面是一个示例代码，用于计算点击量排名前3名的用户：

DataStream<UserBehavior> userBehaviorStream = ...;

DataStream<UserViewCount> windowedData = userBehaviorStream
    .filter(new FilterFunction<UserBehavior>() {
        @Override
        public boolean filter(UserBehavior userBehavior) throws Exception {
            return userBehavior.getBehavior().equals("pv");
        }
    })
    .keyBy(new KeySelector<UserBehavior, Long>() {
        @Override
        public Long getKey(UserBehavior userBehavior) throws Exception {
            return userBehavior.getItemId();
        }
    })
    .timeWindow(Time.hours(1), Time.minutes(5))
    .aggregate(new CountAgg(), new WindowResultFunction());

DataStream<String> topItems = windowedData
    .keyBy("windowEnd")
    .process(new TopNHotUsers(3))
    .map(new MapFunction<Tuple2<Long, String>, String>() {
        @Override
        public String map(Tuple2<Long, String> value) throws Exception {
            return "窗口结束时间: " + new Timestamp(value.f0) + "\n" + value.f1;
        }
    });

topItems.print();

其中，TopNHotUsers是一个自定义的ProcessFunction，用于计算排名前N的用户。具体实现可以参考以下代码：

public class TopNHotUsers extends KeyedProcessFunction<Long, UserViewCount, Tuple2<Long, String>> {
    private final int topSize;

    public TopNHotUsers(int topSize) {
        this.topSize = topSize;
    }

    private ListState<UserViewCount> itemState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ListStateDescriptor<UserViewCount> itemsStateDesc = new ListStateDescriptor<>("itemState-state", UserViewCount.class);
        itemState = getRuntimeContext().getListState(itemsStateDesc);
    }

    @Override
    public void processElement(UserViewCount userViewCount, Context context, Collector<Tuple2<Long, String>> collector) throws Exception {
        itemState.add(userViewCount);
        context.timerService().registerEventTimeTimer(userViewCount.getWindowEnd() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<Long, String>> out) throws Exception {
        List<UserViewCount> allItems = new ArrayList<>();
        for (UserViewCount item : itemState.get()) {
            allItems.add(item);
        }
        itemState.clear();
        allItems.sort(new Comparator<UserViewCount>() {
            @Override
            public int compare(UserViewCount o1, UserViewCount o2) {
                return (int) (o2.getViewCount() - o1.getViewCount());
            }
        });
        StringBuilder result = new StringBuilder();
        result.append("====================================\n");
        result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n");
        for (int i = 0; i < Math.min(topSize, allItems.size()); i++) {
            UserViewCount currentItem = allItems.get(i);
            result.append("No").append(i + 1).append(":")
                    .append("  商品ID=").append(currentItem.getItemId())
                    .append("  浏览量=").append(currentItem.getViewCount())
                    .append("\n");
        }
        result.append("====================================\n\n");
        out.collect(Tuple2.of(timestamp - 1, result.toString()));
    }
}