写点什么

北京大数据培训 | 电商用户行为分析之实时流量统计

作者:@零度
  • 2022 年 6 月 22 日
  • 本文字数:6752 字

    阅读完需:约 22 分钟

​模块创建和数据准备


在 UserBehaviorAnalysis 下 新 建 一 个 maven module 作 为 子 项 目 , 命 名 为 NetworkFlowAnalysis。在这个子模块中,我们同样并没有引入更多的依赖,所以也不需要改动 pom 文件。


在 src/main/目录下,将默认源文件目录 java 改名为 scala。将 apache 服务器的日志文件 apache.log 复制到资源文件目录 src/main/resources 下,我们将从这里读取数据。


当然,我们也可以仍然用 UserBehavior.csv 作为数据源,这时我们分析的就不是每一次对服务器的访问请求了,而是具体的页面浏览(“pv”)操作。


基于服务器 log 的热门页面浏览量统计


我们现在要实现的模块是 “实时流量统计”。对于一个电商平台而言,用户登录的入口流量、不同页面的访问流量都是值得分析的重要数据,而这些数据,可以简单地从 web 服务器的日志中提取出来_大数据培训


我们在这里先实现“热门页面浏览数”的统计,也就是读取服务器日志中的每一行 log,统计在一段时间内用户访问每一个 url 的次数,然后排序输出显示。


具体做法为:每隔 5 秒,输出最近 10 分钟内访问量最多的前 N 个 URL。可以看出,这个需求与之前“实时热门商品统计”非常类似,所以我们完全可以借鉴此前的代码。


在 src/main/scala 下创建 NetworkFlow.scala 文件,新建一个单例对象。定义样例类 ApacheLogEvent,这是输入的日志数据流;另外还有 UrlViewCount,这是窗口操作统计的输出数据类型。在 main 函数中创建 StreamExecutionEnvironment 并做配置,然后从 apache.log 文件中读取数据,并包装成 ApacheLogEvent 类型。


需要注意的是,原始日志中的时间是“dd/MM/yyyy:HH:mm:ss”的形式,需要定义一个 DateTimeFormat 将其转换为我们需要的时间戳格式:


.map(line => {


val linearray = line.split(" ")


val sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")


val timestamp = sdf.parse(linearray(3)).getTime


ApacheLogEvent(linearray(0), linearray(2), timestamp,


linearray(5), linearray(6))


})


完整代码如下:


NetworkFlowAnalysis/src/main/scala/NetworkFlow.scala


case class ApacheLogEvent(ip: String, userId: String, eventTime: Long, method: String,


url: String)


case class UrlViewCount(url: String, windowEnd: Long, count: Long)


object NetworkFlow{


def main(args: Array[String]): Unit = {


val env = StreamExecutionEnvironment.getExecutionEnvironment


env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


env.setParallelism(1)


val stream = env


// 以 window 下为例,需替换成自己的路径


.readTextFile("YOUR_PATH\resources\apache.log")


.map(line => {


val linearray = line.split(" ")


val simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")


val timestamp = simpleDateFormat.parse(linearray(3)).getTime


ApacheLogEvent(linearray(0), linearray(2), timestamp, linearray(5),


linearray(6))


})


.assignTimestampsAndWatermarks(new


BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent]


(Time.milliseconds(1000)) {


override def extractTimestamp(t: ApacheLogEvent): Long = {


t.eventTime


}


})


.filter( data => {


val pattern = "^((?!\.(css|js)".r


(pattern findFirstIn data.url).nonEmpty


} )


.keyBy("url")


.timeWindow(Time.minutes(10), Time.seconds(5))


.aggregate(new CountAgg(), new WindowResultFunction())


.keyBy(1)


.process(new TopNHotUrls(5))


.print()


env.execute("Network Flow Job")


}


class CountAgg extends AggregateFunction[ApacheLogEvent, Long, Long] {


override def createAccumulator(): Long = 0L


override def add(apacheLogEvent: ApacheLogEvent, acc: Long): Long = acc + 1


override def getResult(acc: Long): Long = acc


override def merge(acc1: Long, acc2: Long): Long = acc1 + acc2


}


class WindowResultFunction extends WindowFunction[Long, UrlViewCount, Tuple,


TimeWindow] {


override def apply(key: Tuple, window: TimeWindow, aggregateResult: Iterable[Long],


collector: Collector[UrlViewCount]) : Unit = {


val url: String = key.asInstanceOf[Tuple1[String]].f0


val count = aggregateResult.iterator.next


collector.collect(UrlViewCount(url, window.getEnd, count))


}


}


class TopNHotUrls(topsize: Int) extends KeyedProcessFunction[Tuple, UrlViewCount,


String] {


private var urlState : ListState[UrlViewCount] = _


override def open(parameters: Configuration): Unit = {


super.open(parameters)


val urlStateDesc = new ListStateDescriptor[UrlViewCount]("urlState-state",


classOf[UrlViewCount])


urlState = getRuntimeContext.getListState(urlStateDesc)


}


override def processElement(input: UrlViewCount, context:


KeyedProcessFunction[Tuple, UrlViewCount, String]#Context, collector:


Collector[String]): Unit = {


// 每条数据都保存到状态中


urlState.add(input)


context.timerService.registerEventTimeTimer(input.windowEnd + 1)


}


override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, UrlViewCount,


String]#OnTimerContext, out: Collector[String]): Unit = {


// 获取收到的所有 URL 访问量


val allUrlViews: ListBuffer[UrlViewCount] = ListBuffer()


import scala.collection.JavaConversions._


for (urlView <- urlState.get) {


allUrlViews += urlView


}


// 提前清除状态中的数据,释放空间


urlState.clear()


// 按照访问量从大到小排序


val sortedUrlViews = allUrlViews.sortBy(_.count)(Ordering.Long.reverse)


.take(topSize)


// 将排名信息格式化成 String, 便于打印


var result: StringBuilder = new StringBuilder


result.append("====================================\n")


result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n")


for (i <- sortedUrlViews.indices) {


val currentUrlView: UrlViewCount = sortedUrlViews(i)


// e.g. No1: URL=/blog/tags/firefox?flav=rss20 流量=55


result.append("No").append(i+1).append(":")


.append(" URL=").append(currentUrlView.url)


.append(" 流量=").append(currentUrlView.count).append("\n")


}


result.append("====================================\n\n")


// 控制输出频率,模拟实时滚动结果


Thread.sleep(1000)


out.collect(result.toString)


}


}


}


基于埋点日志数据的网络流量统计


我们发现,从 web 服务器 log 中得到的 url,往往更多的是请求某个资源地址(/.js、/.css),如果要针对页面进行统计往往还需要进行过滤。而在实际电商应用中,相比每个单独页面的访问量,我们可能更加关心整个电商网站的网络流量。


这个指标,除了合并之前每个页面的统计结果之外,还可以通过统计埋点日志数据中的“pv”行为来得到_大数据视频


网站总浏览量(PV)的统计


衡量网站流量一个最简单的指标,就是网站的页面浏览量(Page View,PV)。用户每次打开一个页面便记录 1 次 PV,多次打开同一页面则浏览量累计。一般来说,PV 与来访者的数量成正比,但是 PV 并不直接决定页面的真实来访者数量,如同一个来访者通过不断的刷新页面,也可以制造出非常高的 PV。


我们知道,用户浏览页面时,会从浏览器向网络服务器发出一个请求(Request),网络服务器接到这个请求后,会将该请求对应的一个网页(Page)发送给浏览器,从而产生了一个 PV。所以我们的统计方法,可以是从 web 服务器的日志中去提取对应的页面访问然后统计,就向上一节中的做法一样;也可以直接从埋点日志中提取用户发来的页面请求,从而统计出总浏览量。


所以,接下来我们用 UserBehavior.csv 作为数据源,实现一个网站总浏览量的统计。我们可以设置滚动时间窗口,实时统计每小时内的网站 PV。


在 src/main/scala 下创建 PageView.scala 文件,具体代码如下:


NetworkFlowAnalysis/src/main/scala/PageView.scala


case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String,


timestamp: Long)


object PageView {


def main(args: Array[String]): Unit = {


val resourcesPath = getClass.getResource("/UserBehaviorTest.csv")


val env = StreamExecutionEnvironment.getExecutionEnvironment


env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


env.setParallelism(1)


val stream = env.readTextFile(resourcesPath.getPath)


.map(data => {


val dataArray = data.split(",")


UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt,


dataArray(3), dataArray(4).toLong)


})


.assignAscendingTimestamps(_.timestamp * 1000)


.filter(_.behavior == "pv")


.map(x => ("pv", 1))


.keyBy(_._1)


.timeWindow(Time.seconds(60 * 60))


.sum(1)


.print()


env.execute("Page View Job")


}


}


网站独立访客数(UV)的统计


在上节的例子中,我们统计的是所有用户对页面的所有浏览行为,也就是说,同一用户的浏览行为会被重复统计。而在实际应用中,我们往往还会关注,在一段时间内到底有多少不同的用户访问了网站。


另外一个统计流量的重要指标是网站的独立访客数(Unique Visitor,UV)。UV 指的是一段时间(比如一小时)内访问网站的总人数,1 天内同一访客的多次访问只记录为一个访客。通过 IP 和 cookie 一般是判断 UV 值的两种方式。当客户端第一次访问某个网站服务器的时候,网站服务器会给这个客户端的电脑发出一个 Cookie,通常放在这个客户端电脑的 C 盘当中。在这个 Cookie 中会分配一个独一无二的编号,这其中会记录一些访问服务器的信息,如访问时间,访问了哪些页面等等。当你下次再访问这个服务器的时候,服务器就可以直接从你的电脑中找到上一次放进去的 Cookie 文件,并且对其进行一些更新,但那个独一无二的编号是不会变的。


当然,对于 UserBehavior 数据源来说,我们直接可以根据 userId 来区分不同的用户。


在 src/main/scala 下创建 UniqueVisitor.scala 文件,具体代码如下:


NetworkFlowAnalysis/src/main/scala/UniqueVisitor.scala


case class UvCount(windowEnd: Long, count: Long)


object UniqueVisitor {


def main(args: Array[String]): Unit = {


val resourcesPath = getClass.getResource("/UserBehaviorTest.csv")


val env = StreamExecutionEnvironment.getExecutionEnvironment


env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


env.setParallelism(1)


val stream = env


.readTextFile(resourcesPath.getPath)


.map(line => {


val linearray = line.split(",")


UserBehavior(linearray(0).toLong, linearray(1).toLong, linearray(2).toInt,


linearray(3), linearray(4).toLong)


})


.assignAscendingTimestamps(_.timestamp * 1000)


.filter(_.behavior == "pv")


.timeWindowAll(Time.seconds(60 * 60))


.apply(new UvCountByWindow())


.print()


env.execute("Unique Visitor Job")


}


}


class UvCountByWindow extends AllWindowFunction[UserBehavior, UvCount, TimeWindow] {


override def apply(window: TimeWindow,


input: Iterable[UserBehavior],


out: Collector[UvCount]): Unit = {


val s: collection.mutable.Set[Long] = collection.mutable.Set()


var idSet = SetLong


for ( userBehavior <- input) {


idSet += userBehavior.userId


}


out.collect(UvCount(window.getEnd, idSet.size))


}


}


使用布隆过滤器的 UV 统计


在上节的例子中,我们把所有数据的 userId 都存在了窗口计算的状态里,在窗口收集数据的过程中,状态会不断增大。一般情况下,只要不超出内存的承受范围,这种做法也没什么问题;但如果我们遇到的数据量很大呢?


把所有数据暂存放到内存里,显然不是一个好注意。我们会想到,可以利用 redis 这种内存级 k-v 数据库,为我们做一个缓存。但如果我们遇到的情况非常极端,数据大到惊人呢?比如上亿级的用户,要去重计算 UV。


如果放到 redis 中,亿级的用户 id(每个 20 字节左右的话)可能需要几 G 甚至几十 G 的空间来存储。当然放到 redis 中,用集群进行扩展也不是不可以,但明显代价太大了。


一个更好的想法是,其实我们不需要完整地存储用户 ID 的信息,只要知道他在不在就行了。所以其实我们可以进行压缩处理,用一位(bit)就可以表示一个用户的状态。这个思想的具体实现就是布隆过滤器(Bloom Filter)。


本质上布隆过滤器是一种数据结构,比较巧妙的概率型数据结构(probabilisticdata structure),特点是高效地插入和查询,可以用来告诉你 “某样东西一定不存在或者可能存在”。


它本身是一个很长的二进制向量,既然是二进制的向量,那么显而易见的,存放的不是 0,就是 1。相比于传统的 List、Set、Map 等数据结构,它更高效、占用空间更少,但是缺点是其返回的结果是概率性的,而不是确切的。


我们的目标就是,利用某种方法(一般是 Hash 函数)把每个数据,对应到一个位图的某一位上去;如果数据存在,那一位就是 1,不存在则为 0。


接下来我们就来具体实现一下。


注意这里我们用到了 redis 连接存取数据,所以需要加入 redis 客户端的依赖:


<dependencies>


<dependency>


<groupId>redis.clients</groupId>


<artifactId>jedis</artifactId>


<version>2.8.1</version>


</dependency>


</dependencies>


在 src/main/scala 下创建 UniqueVisitor.scala 文件,具体代码如下:


NetworkFlowAnalysis/src/main/scala/UvWithBloom.scala


object UvWithBloomFilter {


def main(args: Array[String]): Unit = {


val env = StreamExecutionEnvironment.getExecutionEnvironment


env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


env.setParallelism(1)


val resourcesPath = getClass.getResource("/UserBehaviorTest.csv")


val stream = env


.readTextFile(resourcesPath.getPath)


.map(data => {


val dataArray = data.split(",")


UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt,


dataArray(3), dataArray(4).toLong)


})


.assignAscendingTimestamps(_.timestamp * 1000)


.filter(_.behavior == "pv")


.map(data => ("dummyKey", data.userId))


.keyBy(_._1)


.timeWindow(Time.seconds(60 * 60))


.trigger(new MyTrigger()) // 自定义窗口触发规则


.process(new UvCountWithBloom()) // 自定义窗口处理规则


stream.print()


env.execute("Unique Visitor with bloom Job")


}


}


// 自定义触发器


class MyTrigger() extends Trigger[(String, Long), TimeWindow] {


override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext):


TriggerResult = {


TriggerResult.CONTINUE


}


override def onProcessingTime(time: Long, window: TimeWindow, ctx:


Trigger.TriggerContext): TriggerResult = {


TriggerResult.CONTINUE


}


override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = {


}


override def onElement(element: (String, Long), timestamp: Long, window: TimeWindow,


ctx: Trigger.TriggerContext): TriggerResult = {


// 每来一条数据,就触发窗口操作并清空


TriggerResult.FIRE_AND_PURGE


}


}


// 自定义窗口处理函数


class UvCountWithBloom() extends ProcessWindowFunction[(String, Long), UvCount, String,


TimeWindow] {


// 创建 redis 连接


lazy val jedis = new Jedis("localhost", 6379)


lazy val bloom = new Bloom(1 << 29)


override def process(key: String, context: Context, elements: Iterable[(String, Long)],


out: Collector[UvCount]): Unit = {


val storeKey = context.window.getEnd.toString


var count = 0L


if (jedis.hget("count", storeKey) != null) {


count = jedis.hget("count", storeKey).toLong


}


val userId = elements.last._2.toString


val offset = bloom.hash(userId, 61)


val isExist = jedis.getbit(storeKey, offset)


if (!isExist) {


jedis.setbit(storeKey, offset, true)


jedis.hset("count", storeKey, (count + 1).toString)


out.collect(UvCount(storeKey.toLong, count + 1))


} else {


out.collect(UvCount(storeKey.toLong, count))


}


}


}


// 定义一个布隆过滤器


class Bloom(size: Long) extends Serializable {


private val cap = size


def hash(value: String, seed: Int): Long = {


var result = 0


for (i <- 0 until value.length) {


// 最简单的 hash 算法,每一位字符的 ascii 码值,乘以 seed 之后,做叠加


result = result * seed + value.charAt(i)


}


(cap - 1) & result


}


}


用户头像

@零度

关注

关注尚硅谷,轻松学IT 2021.11.23 加入

IT培训 www.atguigu.com

评论

发布
暂无评论
北京大数据培训 | 电商用户行为分析之实时流量统计_大数据开发_@零度_InfoQ写作社区