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);