Flink 计算 PV UV

{"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)));
		env.addSource(consumer).filter(new FilterFunction<String>() {			@Override			public boolean filter(String value) throws Exception {				return true;			}		}).windowAll(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))				.allowedLateness(Time.minutes(1))//				.trigger(CountTrigger.of(5))// 其实多个 trigger 就是下一个 trigger 覆盖上一个 trigger				//用 event time 可能会导致 window 延迟触发，最好的解决办法是在 processingTime 的基础上添加对窗口的判断				// watermark 不会回退,所以如果消息早到的话( 乱序了，该相对来说晚到的消息早到了)，可能会导致窗口延迟触发				// 夸张一点的话，窗口不触发了，直到有大于等于 watermark + triggerTime 的消息到达				// ContinuousProcessingTimeTrigger 一样				.trigger(ContinuousEventTimeTrigger.of(Time.seconds(30)))				//追历史数据的时候会有问题，可能历史数据不足 10s 就全部消费完毕，导致窗口不会被触发而被跳过，消费同理//				.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))				//处理完毕后将 window state 中的数据清除掉				// 其实完全可以通过自定义 trigger 来达到 clear windowState 的目的 (Purge)				.evictor(TimeEvictor.of(Time.seconds(0), true))				.process(new ProcessAllWindowFunction<String, String, TimeWindow>() {					private JedisCluster jedisCluster;					private MapState<String, String> courierInfoMapState;					private MapStateDescriptor<String, String> mapStateDescriptor;					private MapStateDescriptor<String, Long> mapStateUVDescriptor;					private MapState<String, Long> courierInfoUVMapState;					private MapStateDescriptor<String, Long> mapStatePVDescriptor;					private MapState<String, Long> courierInfoPVMapState;					private String beforeDay = "";					private String currentDay = "";
					@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();
						mapStateDescriptor =								new MapStateDescriptor<String, String>("courierInfos", TypeInformation.of(String.class), TypeInformation.of(String.class));						mapStateDescriptor.enableTimeToLive(ttlConfig);						courierInfoMapState = getRuntimeContext().getMapState(mapStateDescriptor);
						mapStateUVDescriptor =								new MapStateDescriptor<String, Long>("courierUVStateDesc", TypeInformation.of(String.class), TypeInformation.of(Long.class));						mapStateUVDescriptor.enableTimeToLive(ttlConfig);						courierInfoUVMapState = getRuntimeContext().getMapState(mapStateUVDescriptor);
						mapStatePVDescriptor =								new MapStateDescriptor<String, Long>("courierPVStateDesc", TypeInformation.of(String.class), TypeInformation.of(Long.class));						mapStatePVDescriptor.enableTimeToLive(ttlConfig);						courierInfoPVMapState = getRuntimeContext().getMapState(mapStatePVDescriptor);

						jedisCluster = RedisUtil.getJedisCluster(redisHp);					}
					@Override					public void close() throws Exception {						RedisUtil.closeConn(jedisCluster);					}
					@Override					public void process(Context context, Iterable<String> elements, Collector<String> out) throws Exception {						Iterator<String> iterator = elements.iterator();						TimeWindow window = context.window();						System.out.println(" window = "								+ DateUtils.millisecondsToDateStr(window.getStart(), "yyyy-MM-dd HH:mm:ss")								+ "-" + DateUtils.millisecondsToDateStr(window.getEnd(), "yyyy-MM-dd HH:mm:ss"));						while (iterator.hasNext()) {							Map<String, Object> map = Json2Others.json2map(iterator.next());							String courierId = map.get("courierId").toString();							String day = map.get("locTime").toString().split(" ")[0].replace("-", "");							if (courierInfoPVMapState.contains(day)) {								courierInfoPVMapState.put(day, courierInfoPVMapState.get(day) + 1);							} else {								courierInfoPVMapState.put(day, 1L);							}							if (!courierInfoMapState.contains(day + "-" + courierId)) {								if (courierInfoUVMapState.contains(day)) {									courierInfoUVMapState.put(day, courierInfoUVMapState.get(day) + 1);								} else {									courierInfoUVMapState.put(day, 1L);								}								courierInfoMapState.put(day + "-" + courierId, "");							}							currentDay = day;						}
						HashMap<String, String> map = new HashMap<String, String>();						if (currentDay.equals(beforeDay)) {							map.put(currentDay + "-pv", courierInfoPVMapState.get(currentDay).toString());							map.put(currentDay + "-uv", courierInfoUVMapState.get(currentDay).toString());
						} else {							map.put(currentDay + "-pv", courierInfoPVMapState.get(currentDay).toString());							map.put(currentDay + "-uv", courierInfoUVMapState.get(currentDay).toString());							//超过25个小时，昨天的数据就不对了							if (!beforeDay.isEmpty()) {								map.put(beforeDay + "-pv", courierInfoPVMapState.get(beforeDay).toString());								map.put(beforeDay + "-uv", courierInfoUVMapState.get(beforeDay).toString());							}						}						map.forEach((k, v) -> {							System.out.println(k + ":" + v);						});						jedisCluster.hmset("test_courier_puv:", map);						jedisCluster.expire("test_courier_puv:", 3 * 24 * 60 * 60);
						beforeDay = currentDay;
					}				});