Flink 计算 TopN

{"locTime":"2020-12-28 12:32:23","courierId":12,"other":"aaa"}

FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(topics, new SimpleStringSchema(), properties);
		FlinkHelp.setOffset(parameter, consumer);
		consumer.assignTimestampsAndWatermarks(
				WatermarkStrategy.<String>forMonotonousTimestamps()
						.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
							@Override
							public long extractTimestamp(String element, long recordTimestamp) {
								String locTime = "";
								try {
									Map<String, Object> map = Json2Others.json2map(element);
									locTime = map.get("locTime").toString();
								} catch (IOException e) {
								}
								LocalDateTime startDateTime =
										LocalDateTime.parse(locTime, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
								long milli = startDateTime.toInstant(OffsetDateTime.now().getOffset()).toEpochMilli();
								return milli;
							}
						}).withIdleness(Duration.ofSeconds(1)));

		SingleOutputStreamOperator<CourierListenInfos> process = env.addSource(consumer).filter(new FilterFunction<String>() {
			@Override
			public boolean filter(String value) throws Exception {
				return true;
			}
		}).keyBy(new KeySelector<String, String>() {
			@Override
			public String getKey(String value) throws Exception {
				Map<String, Object> map = Json2Others.json2map(value);
				String courierId = map.get("courierId").toString();
				String day = map.get("locTime").toString().split(" ")[0].replace("-", "");
				return day + "-" + courierId;
			}
		}).window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
				.allowedLateness(Time.minutes(1))
//				.trigger(CountTrigger.of(5))// 其实多个 trigger 就是下一个 trigger 覆盖上一个 trigger
				.trigger(ContinuousEventTimeTrigger.of(Time.seconds(30)))
				//追历史数据的时候会有问题
//				.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
				//处理完毕后将 window state 中的数据清除掉
				.evictor(TimeEvictor.of(Time.seconds(0), true))
				.process(new ProcessWindowFunction<String, CourierListenInfos, String, TimeWindow>() {
					private JedisCluster jedisCluster;
					private ReducingStateDescriptor<Long> reducingStateDescriptor;
					private ReducingState<Long> listenCount;

					@Override
					public void open(Configuration parameters) throws Exception {
						StateTtlConfig ttlConfig = StateTtlConfig
								.newBuilder(org.apache.flink.api.common.time.Time.hours(25))
								//default，不支持 eventTime 1.12.0
								.setTtlTimeCharacteristic(StateTtlConfig.TtlTimeCharacteristic.ProcessingTime)
								.cleanupInRocksdbCompactFilter(1000)
								.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)//default
								.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
								.build();

						reducingStateDescriptor =
								new ReducingStateDescriptor<Long>("listenCount", new Sum(), TypeInformation.of(Long.class));
						reducingStateDescriptor.enableTimeToLive(ttlConfig);
						listenCount = getRuntimeContext().getReducingState(reducingStateDescriptor);

						jedisCluster = RedisUtil.getJedisCluster(redisHp);
					}

					@Override
					public void close() throws Exception {
						RedisUtil.closeConn(jedisCluster);
					}

					@Override
					public void process(String s, Context context, Iterable<String> elements, Collector<CourierListenInfos> out) throws Exception {
						Iterator<String> iterator = elements.iterator();

						long l = context.currentProcessingTime();
						long watermark = context.currentWatermark();
						TimeWindow window = context.window();

						String endDay = DateUtils.millisecondsToDateStr(window.getEnd(), "yyyyMMdd HH:mm:ss");
						String startDay = DateUtils.millisecondsToDateStr(window.getStart(), "yyyyMMdd HH:mm:ss");

						System.out.println("currentProcessingTime:" + l + " watermark:" + watermark + " windowTime:" + startDay + "-" + endDay);

						while (iterator.hasNext()) {
							iterator.next();
							listenCount.add(1L);
						}

						iterator = elements.iterator();
						Map<String, Object> map = Json2Others.json2map(iterator.next());
						String courierId = map.get("courierId").toString();
						String day = map.get("locTime").toString().split(" ")[0].replace("-", "");
						out.collect(new CourierListenInfos(day, courierId, listenCount.get()));
					}
				});

		process.keyBy(new KeySelector<CourierListenInfos, String>() {
			@Override
			public String getKey(CourierListenInfos value) throws Exception {
				return value.getDay();
			}
		}).process(new KeyedProcessFunction<String, CourierListenInfos, String>() {
			private JedisCluster jedisCluster;
			private MapStateDescriptor<String, Long> mapStateCountDescriptor;
			private MapState<String, Long> courierInfoCountMapState;
			private boolean mucalc = false;

			@Override
			public void open(Configuration parameters) throws Exception {
				StateTtlConfig ttlConfig = StateTtlConfig
						.newBuilder(org.apache.flink.api.common.time.Time.hours(25))
						//default，不支持 eventTime 1.12.0
						.setTtlTimeCharacteristic(StateTtlConfig.TtlTimeCharacteristic.ProcessingTime)
						.cleanupInRocksdbCompactFilter(1000)
						.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)//default
						.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
						.build();

				mapStateCountDescriptor =
						new MapStateDescriptor<String, Long>("courierInfoCountMapState", TypeInformation.of(String.class), TypeInformation.of(Long.class));
				mapStateCountDescriptor.enableTimeToLive(ttlConfig);
				courierInfoCountMapState = getRuntimeContext().getMapState(mapStateCountDescriptor);

				jedisCluster = RedisUtil.getJedisCluster(redisHp);
			}

			@Override
			public void close() throws Exception {
				RedisUtil.closeConn(jedisCluster);
			}

			@Override
			public void processElement(CourierListenInfos value, Context ctx, Collector<String> out) throws Exception {
				courierInfoCountMapState.put(value.getDay() + "#" + value.getCourierId(), value.getListenCount());
//				System.out.println("ctx.timerService().currentWatermark() = " + DateUtils.millisecondsToDateStr(ctx.timerService().currentWatermark(), "yyyyMMdd HH:mm:ss"));
//				System.out.println("ctx.timestamp() = " + DateUtils.millisecondsToDateStr(ctx.timestamp(), "yyyyMMdd HH:mm:ss"));
				ctx.timerService().registerEventTimeTimer(ctx.timerService().currentWatermark() / 1000 + 1000);
			}

			@Override
			public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
				String day = ctx.getCurrentKey();
				PriorityQueue<CourierListenInfos> courierListenInfos = new PriorityQueue<>(new Comparator<CourierListenInfos>() {
					@Override
					public int compare(CourierListenInfos o1, CourierListenInfos o2) {
						return (int) (o1.listenCount - o2.listenCount);
					}
				});

				Iterable<Map.Entry<String, Long>> entries = courierInfoCountMapState.entries();
				for (Map.Entry<String, Long> entry : entries) {
//					System.out.println("entry.getKey() " + entry.getKey());
					String[] split = entry.getKey().split("#", -1);
					courierListenInfos.offer(new CourierListenInfos(split[0], split[1], entry.getValue()));
					if (courierListenInfos.size() > 5) {
						courierListenInfos.poll();
					}
				}

				courierInfoCountMapState.clear();
				String tops = "";
				int size = courierListenInfos.size();
				for (int i = 0; i < size; i++) {
					CourierListenInfos courierListenInfos1 = courierListenInfos.poll();
					System.out.println("courierListenInfos1 " + courierListenInfos1);
					courierInfoCountMapState.put(courierListenInfos1.getDay() + "#" + courierListenInfos1.getCourierId(), courierListenInfos1.listenCount);
					tops = tops + courierListenInfos1.courierId + "#" + courierListenInfos1.listenCount;
					if (i != size - 1) {
						tops += ",";
					}
				}
//				System.out.println("courierListenInfos.poll() = " + tops);
				jedisCluster.hset("test_courier_tops", day + "-top5", tops);
				System.out.println("============");
			}
		}).setParallelism(1);