当前位置： 首页 > news >正文

基于埋点日志数据的网络流量统计 - PV、UV

news 2025/7/4 10:31:33

水善利万物而不争，处众人之所恶，故几于道💦

文章目录

一、 网站总流量数统计 - PV

  1. 需求分析

  2. 代码实现

   方式一

   方式二

   方式三：使用process算子实现

   方式四：使用process算子实现

二、网站独立访客数统计 - UV

  1. 需求分析

  2. 代码实现

一、 网站总流量数统计 - PV

  PV全称 Page View，也就是一个网站的页面浏览量。每当用户进入网站加载或者刷新某个页面时，就会给该网站带来PV量，它往往用来衡量一个网站的流量和用户活跃度。当然了，单个指标并不能全面的反映网站的实际情况，往往需要结合其他的指标进行分析。

1. 需求分析

  埋点采集到的数据格式大概是这个样子（文件已上传资源）

在这里插入图片描述第一个是userId、第二个是itemId、第三个是categoryId、第四个是behavior、第五个是timestamp

所以我们要统计PV的话要先从第四列中筛选出PV，然后再进行累加，求出最终的PV

2. 代码实现

方式一：
  先用 readTextFile()读取文件，然后将读取到的每行数据封装成一个bean对象，再通过 filter过滤出我们需要的PV数据，这时得到的都是封装好的一个个对象没法直接sum，所以通过 map将数据映射为一个个的元组类型（PV，1）然后，使用 keyBy()将他们分到同一个并行度中进行 sum，得出最终的结果。 
public class Flink01_Project_PV {public static void main(String[] args) {Configuration conf = new Configuration();conf.setInteger("rest.port",1000);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);env.setParallelism(2);env.readTextFile("input/UserBehavior.csv")// 将数据封装成 UserBehavior 对象.map(new MapFunction<String, UserBehavior>() {@Overridepublic UserBehavior map(String line) throws Exception {String[] data = line.split(",");return new UserBehavior(Long.valueOf(data[0]),Long.valueOf(data[1]),Integer.valueOf(data[2]),data[3],Long.valueOf(data[4]));}})// 过滤出行为为PV的数据.filter(new FilterFunction<UserBehavior>() {@Overridepublic boolean filter(UserBehavior value) throws Exception {return "pv".equals(value.getBehavior());}})// 因为直接求和的话没法求，所以做一次映射，映射成 (PV,1) 这样的结构.map(new MapFunction<UserBehavior, Tuple2<String,Long>>() {@Overridepublic Tuple2<String, Long> map(UserBehavior value) throws Exception {return Tuple2.of(value.getBehavior(),1L);}})// 然后 将他们通过key进行分组，进入同一个并行度里面 进行求和.keyBy(new KeySelector<Tuple2<String, Long>, String>() {@Overridepublic String getKey(Tuple2<String, Long> value) throws Exception {return value.f0;}})// 进行求和.sum(1).print();try {env.execute();} catch (Exception e) {e.printStackTrace();}}
}

运行结果：
在这里插入图片描述

方式二：

  这种方式省去了方式一的封装对象，其他的思路都一样。

public class Flink01_Project_PV {public static void main(String[] args) {Configuration conf = new Configuration();conf.setInteger("rest.port",1000);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);env.setParallelism(2);env.readTextFile("input/UserBehavior.csv")// 直接过滤出我们想要的数据.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String value) throws Exception {String[] data = value.split(",");return "pv".equals(data[3]);}})// 然后将结构转换为元组类型 (PV,1).map(new MapFunction<String, Tuple2<String,Long>>() {@Overridepublic Tuple2<String, Long> map(String value) throws Exception {String[] data = value.split(",");return Tuple2.of(data[3],1L);}})// 通过key分组.keyBy(new KeySelector<Tuple2<String, Long>, String>() {@Overridepublic String getKey(Tuple2<String, Long> value) throws Exception {return value.f0;}})// 求和.sum(1).print();try {env.execute();} catch (Exception e) {e.printStackTrace();}}
}

运行结果：
在这里插入图片描述

方式三：使用process算子实现

  首先使用readTextFile读取数据，使用map将读取到的数据封装为对象，然后使用keyBy进行分组，最后使用process算子进行求解

  • 为什么要使用keyBy()：目的是让pv数据进入同一个并行度，如果不使用直接process的话，两个并行度里面都有一个sum，结果就不对了
  • 为什么不使用filter过滤呢？因为我们的过滤逻辑是再process里面完成的，所以不用再额外过滤
public class Flink02_Project_PV_process {public static void main(String[] args) {Configuration conf = new Configuration();conf.setInteger("rest.port",1000);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);env.setParallelism(2);env.readTextFile("input/UserBehavior.csv")// 封装成 UserBehavior 对象.map(new MapFunction<String, UserBehavior>() {@Overridepublic UserBehavior map(String line) throws Exception {String[] data = line.split(",");return new UserBehavior(Long.valueOf(data[0]),Long.valueOf(data[1]),Integer.valueOf(data[2]),data[3],Long.valueOf(data[4]));}})// 通过key分组.keyBy(new KeySelector<UserBehavior, String>() {@Overridepublic String getKey(UserBehavior value) throws Exception {return value.getBehavior();}})// 使用proces算子实现 PV 的统计.process(new ProcessFunction<UserBehavior, String>() {// 定义累加变量long sum =0L ;@Overridepublic void processElement(UserBehavior value, Context ctx, Collector<String> out) throws Exception {// 判断用户行为是否是PVif ("pv".equals(value.getBehavior())){// 条件满足 sum+1sum++;// 将结果收集out.collect("pv = "+sum);}}}).print();try {env.execute();} catch (Exception e) {e.printStackTrace();}}
}

运行结果：
在这里插入图片描述

方式四：使用process算子实现

  方式四相比于方式三省去了对象的封装，其他思路一样。

public class Flink02_Project_PV_process {public static void main(String[] args) {Configuration conf = new Configuration();conf.setInteger("rest.port",1000);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);env.setParallelism(2);env.readTextFile("input/UserBehavior.csv")// 通过key分组.keyBy(new KeySelector<String, String>() {@Overridepublic String getKey(String value) throws Exception {return value.split(",")[3];}})// 直接使用process求 PV.process(new ProcessFunction<String, String>() {// 定义累加变量long sum = 0L;@Overridepublic void processElement(String line, Context ctx, Collector<String> out) throws Exception {// 将过来的每行数据切割String[] datas = line.split(",");// 判断是否是我们想要的数据if("pv".equals(datas[3])){// 符合条件，将累加变量+1sum++;// 收集结果out.collect("pv = "+sum);}}}).print();try {env.execute();} catch (Exception e) {e.printStackTrace();}}
}

运行结果：
在这里插入图片描述

二、网站独立访客数统计 - UV

  UV全称 Unique Visitor，也就是独立访客数。在PV中，我们统计的是所有用户对所有页面的浏览行为，也就是同一个用户的浏览行为会被重复统计。实际上我们关注的是在某一特定范围内（一天、一周或者一个月）内访问该网站的用户数，也就是每个访客只计算一次。它能从侧面反映出该网站的受欢迎程度和用户规模的大小。

1. 需求分析

  要统计UV量的话，只需要对全量的PV，使用userId去重，然后就能得到独立访客数了。

2. 代码实现

  先filter过滤出PV数据，然后通过keyBy将PV分到同一组，然后使用process进行处理，处理方法是：用set集合存放userId，如果下一个userId可以加入该集合说明是一个新的独立访客，则收集当前集合的大小，若加入失败，说明集合中已经存在该userId，也就是不是一个新的独立访客，也就不做处理了。

public class Flink03_Project_UV {public static void main(String[] args) {Configuration conf = new Configuration();conf.setInteger("rest.port",1000);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);env.setParallelism(2);env.readTextFile("input/UserBehavior.csv")// 过滤出 PV 数据.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String value) throws Exception {return "pv".equals(value.split(",")[3]);}})// 将PV的数据分到同一个组里面.keyBy(new KeySelector<String, String>() {@Overridepublic String getKey(String value) throws Exception {return value.split(",")[3];}})// 对同一组里面的数据进行处理.process(new ProcessFunction<String, String>() {// 存放 userId 的容器，回自动对数据进行去重，最后直接拿它的大小就知道UV了Set<Long> userIdSet = new HashSet<>();@Overridepublic void processElement(String value, Context ctx, Collector<String> out) throws Exception {Long userId = Long.valueOf(value.split(",")[0]);// 向set中添加userId，判断是否添加成功if (userIdSet.add(userId)) {// 添加成功的话，说明是一个新的独立访客，收集到此时容器大小out.collect("UV = "+ userIdSet.size());}}}).print();try {env.execute();} catch (Exception e) {e.printStackTrace();}}
}

运行结果：
在这里插入图片描述

http://www.lryc.cn/news/106884.html

相关文章：

  • cuda入门demo（2）——最基础的二方向sobel
  • 软件外包开发的后台开发语言
  • 自动驾驶感知系统-全球卫星定位系统
  • 数据结构 | 基本数据结构——队列
  • QT在label上透明绘图（二）
  • 微信小程序使用editor富文本编辑器 以及回显 全屏弹窗的模式
  • 在CSDN学Golang场景化解决方案（基于gin框架的web开发脚手架）
  • 关于Express 5
  • ftrace 原理详细分析
  • UWB定位技术和蓝牙AOA有哪些不同？-高精度室内定位技术对比
  • 【RabbitMQ】golang客户端教程2——工作队列
  • 芯旺微冲刺IPO，车规级MCU竞争白热化下的“隐忧”凸显
  • HTML ＜s＞ 标签
  • 微信小程序 - scroll-view组件之上拉加载下拉刷新（解决上拉加载不触发）
  • rust usize与i64怎么比较大小？
  • 电脑更新win10黑屏解决方法
  • STM32入门——外部中断
  • 【计算机网络】NAT及Bridge介绍
  • 封装动态SQL的插件
  • C# Microsoft消息队列服务器的使用 MSMQ
  • Kafka3.0.0版本——生产者如何提高吞吐量
  • js精度丢失的问题
  • C++ 编译预处理
  • 备战秋招 | 笔试强化22
  • LeetCode ACM模式——哈希表篇（二）
  • hadoop 3.1.3集群搭建 ubuntu20
  • 备忘录模式——撤销功能的实现
  • Golang 函数参数的传递方式 值传递，引用传递
  • K8s影响Pod调度和Deployment
  • 透明代理和不透明代理