Flink-电商用户行为分析(网站独立访客数(UV)的统计)

╰+哭是因爲堅強的太久メ 2023-02-23 14:28 25阅读 0赞

**`数据`**  
链接：https://pan.baidu.com/s/1InfWoNYUeV1KYyvFS1aXuA  
提取码：z3p4  
统计流量的重要指标是网站的独立访客数（Unique Visitor，UV）。UV指的是一段时间（比如一小时）内访问网站的总人数，1 天内同一访客的多次访问只记录为一个访客。通过 IP 和 cookie 一般是判断 UV 值的两种方式。当客户端第一次访问某个网站服务器的时候，网站服务器会给这个客户端的电脑发出一个 Cookie，通常放在这个客户端电脑的C盘当中。在这个 Cookie中会分配一个独一无二的编号，这其中会记录一些访问服务器的信息，如访问时间，访问了哪些页面等等。当你下次再访问这个服务器的时候，服务器就可以直接从你的电脑中找到上一次放进去的Cookie 文件，并且对其进行一些更新，但那个独一无二的编号是不会变的。 当然，对于 UserBehavior 数据源来说，我们直接可以根据 userId 来区分不同的用户。  
**`代码实现`**

import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.scala.function.AllWindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    //定义一个输入数据的样列类
    case class UserBehavior(UserId:Long,itemId:Long,categoryId:Int,behavior:String,timestamp:Long)
    //定义一个输入数据的样列类
    case class UvCount(windowEnd:Long,uvCount:Long)
    
    object UniqueVisitor { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        //使用相对路径
        val resource = getClass.getResource("/UserBehavior.csv")
        val dataStream = env.readTextFile(resource.getPath)
          .map(data => { 
            val dataArray = data.split(",")
            UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim,
              dataArray(4).trim.toLong)
          }).assignAscendingTimestamps(_.timestamp * 1000L)
          .filter(_.behavior == "pv") //只统计pv操作
          //不做keyby直接timeWindowAll
          .timeWindowAll(Time.hours(1))
            .apply(new UvCountByWindow())
    
    
        dataStream.print("pv count")
    
        env.execute("page view job")
      }
    }
    class UvCountByWindow() extends AllWindowFunction[UserBehavior,UvCount,TimeWindow]{ 
      override def apply(window: TimeWindow, input: Iterable[UserBehavior],
                         out: Collector[UvCount]): Unit = { 
        //定义一个scala set 用于保存所有的数据UserId并去重
        var idSet = Set[Long]()
        //把当前窗口所有数据的Id收集到set中，最后输出set的大小
        for (userBehavior <- input){ 
          idSet  += userBehavior.UserId
        }
        out.collect(UvCount(window.getEnd,idSet.size))
      }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
我们需要进一步改进，UV增量去重，因为我们想要来一条数据处理一条，  
代码实现

import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.scala.function.AllWindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    //定义一个输入数据的样列类
    case class UserBehavior(UserId:Long,itemId:Long,categoryId:Int,behavior:String,timestamp:Long)
    //定义一个输入数据的样列类
    case class UvCount(windowEnd:Long,uvCount:Long)
    
    object UniqueVisitor { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        //使用相对路径
        val resource = getClass.getResource("/UserBehavior.csv")
        val dataStream = env.readTextFile(resource.getPath)
          .map(data => { 
            val dataArray = data.split(",")
            UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim,
              dataArray(4).trim.toLong)
          }).assignAscendingTimestamps(_.timestamp * 1000L)
          .filter(_.behavior == "pv") //只统计pv操作
          //不做keyby直接timeWindowAll
          .timeWindowAll(Time.hours(1))
            .aggregate(new UvCountAgg(),new UvCountResult())
    
        dataStream.print("pv count")
        env.execute("page view job")
      }
    }
    //自定义增量聚合函数，需要定义一个Set作为累加状态
    class UvCountAgg() extends AggregateFunction[UserBehavior,Set[Long],Long]{ 
      override def createAccumulator(): Set[Long] = Set[Long]()
    
      override def add(value: UserBehavior, accumulator: Set[Long]): Set[Long] = accumulator + value.UserId
    
      override def getResult(accumulator: Set[Long]): Long = accumulator.size
    
      override def merge(a: Set[Long], b: Set[Long]): Set[Long] = a ++ b
    }
    //自定义窗口函数，添加window信息包装成样例类
    class UvCountResult() extends AllWindowFunction[Long,UvCount,TimeWindow]{ 
      override def apply(window: TimeWindow, input: Iterable[Long], out: Collector[UvCount]): Unit = { 
        out.collect(UvCount(window.getEnd,input.head))
      }
    }

这样可能会报错  
[https://blog.csdn.net/qq\_46548855/article/details/109345958][https_blog.csdn.net_qq_46548855_article_details_109345958]  
但是这样还是有bug！  
如果有1亿个用户 一个用户100B，那么10^10B，相当于10GB  
如果使用布隆过滤器的话，一个用户的100B会转化位1bit，那么就是10^8bit  
10^8bit/1024B/1024KB=100bitM/8bit约等于10MB  
为了防止hash碰撞，我们要扩充10倍，大于要100MB  
![在这里插入图片描述][20201029204022774.png_pic_center]  
**那么我们要使用布隆过滤器**  
我们把所有数据的userId都存在了窗口计算的状态里，在窗口收集数据的过程中，状态会不断增大。一般情况下，只要不超出内存的承受范围，这种做法也没什么问题；但如果我们遇到的数据量很大呢？  
把所有数据暂存放到内存里，显然不是一个好注意。我们会想到，可以利用redis这种内存级k-v数据库，为我们做一个缓存。但如果我们遇到的情况非常极端，数据大到惊人呢？比如上亿级的用户，要去重计算UV。  
如果放到redis中，亿级的用户id（每个20字节左右的话）可能需要几G甚至几十G的空间来存储。当然放到redis中，用集群进行扩展也不是不可以，但明显代价太大了。  
一个更好的想法是，其实我们不需要完整地存储用户ID的信息，只要知道他在不在就行了。所以其实我们可以进行压缩处理，用一位（bit）就可以表示一个用户的状态。这个思想的具体实现就是布隆过滤器（Bloom Filter）。  
本质上布隆过滤器是一种数据结构，比较巧妙的概率型数据结构（probabilistic data structure），特点是高效地插入和查询，可以用来告诉你 “某样东西一定不存在或者可能存在”。  
它本身是一个很长的二进制向量，既然是二进制的向量，那么显而易见的，存放的不是0，就是1。相比于传统的 List、Set、Map 等数据结构，它更高效、占用空间更少，但是缺点是其返回的结果是概率性的，而不是确切的。  
我们的目标就是，利用某种方法（一般是Hash函数）把每个数据，对应到一个位图的某一位上去；如果数据存在，那一位就是1，不存在则为0。  
接下来我们就来具体实现一下。  
注意这里我们用到了redis连接存取数据，所以需要加入redis客户端的依赖：

<dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.8.1</version>
        </dependency>

代码实现

import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.triggers.{ Trigger, TriggerResult}
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    import redis.clients.jedis.Jedis
    
    object UvWithBoomFilter { 
      def main(args: Array[String]): Unit = { 
        // 创建一个流处理执行环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        // 从文件读取数据
        val inputStream: DataStream[String] = env.readTextFile("D:\\idea\\UserBehaviorAnalysis\\NetworkFlowAnalysis\\src\\main\\resources\\UserBehavior.csv")
    
        // 将数据转换成样例类类型，并且提取timestamp定义watermark
        val dataStream: DataStream[UserBehavior] = inputStream
          .map( data => { 
            val dataArray = data.split(",")
            UserBehavior( dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong )
          } )
          .assignAscendingTimestamps(_.timestamp * 1000L)
    
        // 分配key，包装成二元组开创聚合
        val uvStream: DataStream[UvCount] = dataStream
          .filter(_.behavior == "pv")
          .map( data => ("uv", data.UserId) )
          .keyBy(_._1)
          .timeWindow(Time.hours(1))
          .trigger(new MyTrigger())    // 自定义Trigger
          .process( new UvCountResultWithBloomFilter() )
    
        uvStream.print()
    
        env.execute("uv 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
    }
    
    // 自定义ProcessWindowFunction，把当前数据进行处理，位图保存在redis中
    class UvCountResultWithBloomFilter() extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow]{ 
      var jedis: Jedis = _
      var bloom: Bloom = _
    
      override def open(parameters: Configuration): Unit = { 
        jedis = new Jedis("hadoop12", 6379)
        // 位图大小10亿个位，也就是2^30，占用128MB
        bloom = new Bloom(1<<30)
      }
    
      // 每来一个数据，主要是要用布隆过滤器判断redis位图中对应位置是否为1
      override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = { 
        // bitmap用当前窗口的end作为key，保存到redis里，（windowEnd，bitmap）
        val storedKey = context.window.getEnd.toString
    
        // 我们把每个窗口的uv count值，作为状态也存入redis中，存成一张叫做countMap的表
        val countMap = "countMap"
        // 先获取当前的count值
        var count = 0L
        if( jedis.hget(countMap, storedKey) != null )
          count = jedis.hget(countMap, storedKey).toLong
    
        // 取userId，计算hash值，判断是否在位图中
        val userId = elements.last._2.toString
        val offset = bloom.hash(userId, 61)
        val isExist = jedis.getbit( storedKey, offset )
    
        // 如果不存在，那么就将对应位置置1，count加1；如果存在，不做操作
        if( !isExist ){ 
          jedis.setbit( storedKey, offset, true )
          jedis.hset( countMap, storedKey, (count + 1).toString )
        }
      }
    }
    
    // 自定义一个布隆过滤器
    class Bloom(size: Long) extends Serializable{ 
      // 定义位图的大小，应该是2的整次幂
      private val cap = size
    
      // 实现一个hash函数
      def hash(str: String, seed: Int): Long = { 
        var result = 0
        for( i <- 0 until str.length ){ 
          result = result * seed + str.charAt(i)
        }
        // 返回一个在cap范围内的一个值
        (cap - 1) & result
      }
    }

解释计算Trigger里面TriggerResult的一些参数

fire处不触发计算
    purge清不清空状态
    
    CONTINUE(false, false)
    FIRE_AND_PURGE(true, true)
    FIRE(true, false)
    PURGE(false, true)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200707205435352.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_qq_46548855_article_details_109345958]: https://blog.csdn.net/qq_46548855/article/details/109345958
[20201029204022774.png_pic_center]: https://img-blog.csdnimg.cn/20201029204022774.png#pic_center