PV(访问量)：即Page View, 即页面浏览量或点击量，用户每次刷新即被计算一次。
UV(独立访客)：即Unique Visitor,访问您网站的一台电脑客户端为一个访客。00:00-24:00内相同的客户端只被计算一次。
计算网站App的实时pv和uv，是很常见的统计需求，这里提供通用的计算方法，不同的业务需求只需要小改即可拿来即用。
需求
利用Flink实时统计，从0点到当前的pv、uv。
一、需求分析
从Kafka发送过来的数据含有：时间戳、时间、维度、用户id，需要从不同维度统计从0点到当前时间的pv和uv，第二天0点重新开始计数第二天的。
二、技术方案
Kafka数据可能会有延迟乱序，这里引入watermark;
通过keyBy分流进不同的滚动window，每个窗口内计算pv、uv;
由于需要保存一天的状态，process里面使用ValueState保存pv、uv;
使用BitMap类型ValueState，占内存很小，引入支持bitmap的依赖;
保存状态需要设置ttl过期时间，第二天把第一天的过期，避免内存占用过大。
三、数据准备
这里假设是用户订单数据，数据格式如下：
- {"time":"2021-10-3122:00:01","timestamp":"1635228001","product":"苹果手机","uid":255420}
- {"time":"2021-10-3122:00:02","timestamp":"1635228001","product":"MacBookPro","uid":255421}
四、代码实现
整个工程代码截图如下(抹去了一些不方便公开的信息)：
pvuv-project
1. 环境
kafka：1.0.0;
Flink：1.11.0;
2. 发送测试数据
首先发送数据到kafka测试集群，maven依赖：
- org.apache.kafka
- kafka-clients
- 2.4.1
2.4.1
发送代码：
- importcom.alibaba.fastjson.JSON;
- importcom.alibaba.fastjson.JSONObject;
- importjodd.util.ThreadUtil;
- importorg.apache.commons.lang3.StringUtils;
- importorg.junit.Test;
- importjava.io.*;
- publicclassSendDataToKafka{
- @Test
- publicvoidsendData()throwsIOException{
- Stringinpath="E:\\我的文件\\click.txt";
- Stringtopic="click_test";
- intcnt=0;
- Stringline;
- InputStreaminputStream=newFileInputStream(inpath);
- Readerreader=newInputStreamReader(inputStream);
- LineNumberReaderlnr=newLineNumberReader(reader);
- while((line=lnr.readLine())!=null){
- //这里的KafkaUtil是个生产者、消费者工具类，可以自行实现
- KafkaUtil.sendDataToKafka(topic,String.valueOf(cnt),line);
- cnt=cnt+1;
- ThreadUtil.sleep(100);
- }
- }
- }
3. 主要程序
先定义个pojo：
- @NoArgsConstructor
- @AllArgsConstructor
- @Data
- @ToString
- publicclassUserClickModel{
- privateStringdate;
- privateStringproduct;
- privateintuid;
- privateintpv;
- privateintuv;
- }
接着就是使用Flink消费kafka，指定Watermark，通过KeyBy分流，进入滚动窗口函数通过状态保存pv和uv。
- publicclassUserClickMain{
- privatestaticfinalMapconfig=Configuration.initConfig("commons.xml");
- publicstaticvoidmain(String[]args)throwsException{
- //初始化环境，配置相关属性
- StreamExecutionEnvironmentsenv=StreamExecutionEnvironment.getExecutionEnvironment();
- senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- senv.enableCheckpointing(5000,CheckpointingMode.EXACTLY_ONCE);
- senv.setStateBackend(newFsStateBackend("hdfs://bigdata/flink/checkpoints/userClick"));
- //读取kafka
- PropertieskafkaProps=newProperties();
- kafkaProps.setProperty("bootstrap.servers",config.get("kafka-ipport"));
- kafkaProps.setProperty("group.id",config.get("kafka-groupid"));
- //kafkaProps.setProperty("auto.offset.reset","earliest");
- //watrmark允许数据延迟时间
- longmaxOutOfOrderness=5*1000L;
- SingleOutputStreamOperatordataStream=senv.addSource(
- newFlinkKafkaConsumer<>(
- config.get("kafka-topic"),
- newSimpleStringSchema(),
- kafkaProps
- ))
- //设置watermark
- .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(maxOutOfOrderness))
- .withTimestampAssigner((element,recordTimestamp)->{
- //时间戳须为毫秒
- returnLong.valueOf(JSON.parseObject(element).getString("timestamp"))*1000;
- })).map(newFCClickMapFunction()).returns(TypeInformation.of(newTypeHint(){
- }));
- //按照(date,product)分组
- dataStream.keyBy(newKeySelector
如果您对该产品感兴趣，请填写办理（客服微信：xiaoxiongyidong）
发表评论
◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。