基于flink的电商用户行为数据分析【3】| 实时流量统计

大数据梦想家

发布于 2021-01-27 16:49:10

2.1K0

发布于 2021-01-27 16:49:10

文章被收录于专栏：大数据成长之路大数据成长之路

前言

在上一期内容中，菌哥已经为大家介绍了实时热门商品统计模块的功能开发的过程(?基于flink的电商用户行为数据分析【2】| 实时热门商品统计)。本期文章，我们要学习的是实时流量统计模块的开发过程。

模块创建和数据准备

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

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

代码实现

我们现在要实现的模块是 “实时流量统计”。对于一个电商平台而言，用户登录的入口流量、不同页面的访问流量都是值得分析的重要数据，而这些数据，可以简单地从web服务器的日志中提取出来。我们在这里实现最基本的“页面浏览数”的统计，也就是读取服务器日志中的每一行log，统计在一段时间内用户访问url的次数。

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

具体分析如下：

热门页面

基本需求 – 从 web 服务器的日志中，统计实时的热门访问页面 – 统计每分钟的ip访问量，取出访问量最大的5个地址，每5秒更新一次
解决思路 – 将 apache 服务器日志中的时间，转换为时间戳，作为 Event Time – 构建滑动窗口，窗口长度为1分钟，滑动距离为5秒

PV 和 UV

基本需求 – 从埋点日志中，统计实时的 PV 和 UV – 统计每小时的访问量（PV），并且对用户进行去重（UV）
解决思路 – 统计埋点日志中的 pv 行为，利用 Set 数据结构进行去重 – 对于超大规模的数据，可以考虑用布隆过滤器进行去重

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

 // 输入 log 数据样例类
  case class ApacheLogEvent(ip: String, userId: String, eventTime: Long, method: String, url: String)

  // 中间统计结果样例类
  case class UrlViewCount(url: String, windowEnd: Long, count: Long)

需要注意的是，原始日志中的时间是“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))
})

因为后面部分的逻辑可以说与实时商品统计部分的逻辑是一样的，所以这里小菌就不再带着大家一步步去分析了，完整代码如下：

import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/*
 * @Author: Alice菌
 * @Date: 2020/11/23 14:16
 * @Description: 
    电商用户行为数据分析：实时流量统计
    <每隔5秒，输出最近10分钟内访问量最多的前N个URL>
 */
object NetworkFlow {

  // 输入 log 数据样例类
  case class ApacheLogEvent(ip: String, userId: String, eventTime: Long, method: String, url: String)

  // 中间统计结果样例类
  case class UrlViewCount(url: String, windowEnd: Long, count: Long)

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

    // 创建 流处理的 环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置时间语义为 eventTime -- 事件创建的时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 设置任务并行度
    env.setParallelism(1)
    // 读取文件数据
    val stream: DataStream[String] = env.readTextFile("G:\\idea arc\\BIGDATA\\project\\src\\main\\resources\\apache.log")

    // 对 stream 数据进行处理
    stream.map(data => {
      val dataArray: Array[String] = data.split(" ")
      // 因为日志文件中的数据格式是  17/05/2015:10:05:03
      // 所以我们这里用DataFormat对时间进行转换
      val simpleDateFormat: SimpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
      val timestamp: Long = simpleDateFormat.parse(dataArray(3).trim).getTime
      // 将解析的数据存放至我们定义好的样例类中
      ApacheLogEvent(dataArray(0).trim, dataArray(1).trim, timestamp, dataArray(5).trim, dataArray(6).trim)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(60)) {
        override def extractTimestamp(element: ApacheLogEvent): Long = element.eventTime
      })
      // 因为我们需要统计出每种url的出现的次数，故这里将 url 作为 key 进行分组
      .keyBy(_.url)
      // 滑动窗口聚合   -- 每隔5秒，输出最近10分钟内访问量最多的前N个URL
      .timeWindow(Time.minutes(10), Time.seconds(5))
      // 预计算，统计出每个 URL 的访问量
      .aggregate(new CountAgg(),new WindowResult())
      // 根据窗口结束时间进行分组
      .keyBy(_.windowEnd)
      // 输出每个窗口中访问量最多的前5个URL
      .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(value: ApacheLogEvent, accumulator: Long): Long = accumulator + 1

    override def getResult(accumulator: Long): Long = accumulator
 
    override def merge(a: Long, b: Long): Long = a + b

  }
  // 自定义的窗口处理函数
  class WindowResult() extends WindowFunction[Long, UrlViewCount, String, TimeWindow] {

    override def apply(url: String, window: TimeWindow, input: Iterable[Long], out: Collector[UrlViewCount]): Unit = {
      // 输出结果
      out.collect(UrlViewCount(url, window.getEnd, input.iterator.next()))
    }
  }

  // 自定义 process function，实现排序输出
  class TopNHotUrls(nSize: Int) extends KeyedProcessFunction[Long, UrlViewCount, String] {

    // 定义一个状态列表，保存结果
    lazy val urlState: ListState[UrlViewCount] = getRuntimeContext.getListState( new ListStateDescriptor[UrlViewCount]( "urlState", classOf[UrlViewCount] ) )
  
    override def processElement(value: UrlViewCount, ctx: KeyedProcessFunction[Long, UrlViewCount, String]#Context, collector: Collector[String]): Unit = {

       // 将数据添加至 状态 列表中
       urlState.add(value)
      // 根据窗口结束时间windowEnd，设置定时器
       ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)
        
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, UrlViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {

      // 新建一个ListBuffer，用于存放状态列表中的数据
      val  allUrlViews: ListBuffer[UrlViewCount] = new ListBuffer[UrlViewCount]()
      // 获取到状态列表
      val iter: util.Iterator[UrlViewCount] = urlState.get().iterator()
      
      while ( iter.hasNext ) {
        allUrlViews += iter.next()
      }
             
        // 清除状态
        urlState.clear()

        // 按照 count 大小排序
       val sortedUrlViews: ListBuffer[UrlViewCount] = allUrlViews.sortWith(_.count > _.count).take(nSize)
        
       // 格式化成String打印输出
       val 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)
        // 拼接打印结果
        result.append("No").append(i+1).append(":")
          .append("  URL=").append(currentUrlView.url).append(" ")
          .append("  流量=").append(currentUrlView.count).append("\n")
        
    }

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

      // 设置休眠时间
      Thread.sleep(1000)

      // 输出结果
      out.collect(result.toString())

    }
  }

运行效果

为了让小伙伴们更好理解，菌哥基本每行代码都写上了注释，就冲这波细节，还不给安排一波三连?开个玩笑，回到主题上，我们再来讨论一个问题。

实际生产环境中，我们的数据流往往是从Kafka获取到的。如果要让代码更贴近生产实际，我们只需将source更换为Kafka即可：

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

当然，根据实际的需要，我们还可以将Sink指定为Kafka、ES、Redis或其它存储，这里就不一一展开实现了。

参考

https://www.bilibili.com/video/BV1y54y127h2?from=search&seid=5631307517601819264

小结

本期内容主要为大家分享了如何基于flink在电商用户行为分析项目中对实时流量统计模块进行开发的过程，这个跟上一期介绍的实时热门商品统计功能非常类似，对本期内容不太理解的小伙伴可以多研究上一期的精彩内容~下一期我们会介绍项目中恶意登录监控的功能开发，敬请期待！你知道的越多，你不知道的也越多，我是Alice，我们下一期见！

受益的朋友记得三连支持小菌！