Flink-电商用户行为分析(每隔 5 分钟输出最近一小时内点击量最多的前 N 个商品)

「爱情、让人受尽委屈。」 2023-02-23 07:44 50阅读 0赞

**`数据`**  
链接：https://pan.baidu.com/s/1InfWoNYUeV1KYyvFS1aXuA  
提取码：z3p4  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
**`程序主体`**

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    
    //640533,3168550,4719814,pv,1511690399
    //定义输出数据的样例类
    case class UserBehavior(UserID:Long,itemId:Long,categoryId:Int,behavior:String,timestamp:Long)
    //定义窗口聚合结果样例类
    case class ItemViewCount(itemId:Long,windowEnd:Long,count:Long)
    
    object HotItems { 
      def main(args: Array[String]): Unit = { 
        //1.创建执行程序
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        //2.读取数据
        val dataStream = env.readTextFile("D:\\idea\\flinkUser\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")
          .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)
          })
          //*1000 要看是不是秒 秒*1000
          .assignAscendingTimestamps(_.timestamp*1000L)
        
        //4.sink控制台输出
        dataStream.print()
    
        env.execute("hot items job")
      }
    
    }

这里注意，我们需要统计业务时间上的每小时的点击量，所以要基于 EventTime来处理。那么如果让 Flink 按照我们想要的业务时间来处理呢？这里主要有两件事情要做。  
第一件是告诉 Flink 我们现在按照 EventTime 模式进行处理,Flink 默认使用ProcessingTime处理，所以我们要显式设置如下  
第二件事情是指定如何获得业务时间，以及生成 Watermark。Watermark 是用来追踪业务事件的概念，可以理解成 EventTime 世界中的时钟，用来指示当前处理到什么时刻的数据了。由于我们的数据源的数据已经经过整理，没有乱序，即事件的时间戳是单调递增的，所以可以将每条数据的业务时间就当做 Watermark.这里我们用 assignAscendingTimestamps 来实现时间戳的抽取和 Watermark 的生成。  
**`注：真实业务场景一般都是乱序的，所以一般不用assignAscendingTimestamps，而是 使用BoundedOutOfOrdernessTimestampExtractor`**

.assignAscendingTimestamps(_.timestamp * 1000)

这样我们就得到了一个带有时间标记的数据流了，后面就能做一些窗口的操作。  
过滤出点击事件  
在开始窗口操作之前，先回顾下需求“每隔 5 分钟输出过去一小时内点击量最多的前 N 个商品”。由于原始数据中存在点击、购买、收藏、喜欢各种行为的数据，但是我们只需要统计点击量，所以先使用 filter 将点击行为数据过滤出来。  
设置滑动窗口，统计点击量  
由于要每隔 5 分钟统计一次最近一小时每个商品的点击量，所以窗口大小是一小时，每隔 5分钟滑动一次。即分别要统计\[09:00, 10:00), \[09:05, 10:05), \[09:10, 10:10)…等窗口的商品点击量。是一个常见的滑动窗口需求（Sliding Window）  
**`尽量不要写成`** .keyBy(“itemId”) 导致后面写的复杂

.keyBy(_.itemId) 
     .timeWindow(Time.minutes(60), Time.minutes(5)) 
     .aggregate(new CountAgg(), new WindowResultFunction());

我们使用.keyBy(“itemId”)对商品进行分组，使用.timeWindow(Time size, Time slide)对每个商品做滑动窗口（1 小时窗口，5 分钟滑动一次）。然后我们使 用 .aggregate(AggregateFunction af, WindowFunction wf) 做增量的聚合操作，它能使用 AggregateFunction 提 前 聚 合 掉 数 据 ， 减 少 state 的 存 储 压 力 。 较 之 .apply(WindowFunction wf) 会将窗口中的数据都存储下来，最后一起计算要高效 地多。这里的 CountAgg 实现了 AggregateFunction 接口，功能是统计窗口中的条数，  
即遇到一条数据就加一

class CountAgg() extends AggregateFunction[UserBehavior,Long,Long]{ 
      override def createAccumulator(): Long = 0L
    
      override def add(in: UserBehavior, acc: Long): Long = acc + 1
    
      override def getResult(acc: Long): Long = acc
    
      override def merge(acc: Long, acc1: Long): Long = acc + acc1
    }

聚合操作.aggregate(AggregateFunction af, WindowFunction wf)的第二个参数 WindowFunction 将每个 key 每个窗口聚合后的结果带上其他信息进行输出。我们这  
里实现的 WindowResultFunction 将 <主键商品 ID，窗口，点击量 >封装成了  
ItemViewCount 进行输出。

// 商品点击量(窗口操作的输出类型) 
    case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long)

class WindowResult() extends WindowFunction[Long,ItemViewCount,Long,TimeWindow]{ 
      override def apply(key: Long, window: TimeWindow, input: Iterable[Long],
                         out: Collector[ItemViewCount]): Unit = { 
        out.collect(ItemViewCount(key,window.getEnd,input.iterator.next()))
      }
    }

最终的代码

import java.sql.Timestamp
    
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.api.common.state.{ ListState, ListStateDescriptor}
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.api.scala._
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    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
    
    //640533,3168550,4719814,pv,1511690399
    //定义输出数据的样例类
    case class UserBehavior(UserID:Long,itemId:Long,categoryId:Int,behavior:String,timestamp:Long)
    //定义窗口聚合结果样例类
    case class ItemViewCount(itemId:Long,windowEnd:Long,count:Long)
    
    object HotItems { 
      def main(args: Array[String]): Unit = { 
        //1.创建执行程序
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        //2.读取数据
        val dataStream = env.readTextFile("D:\\idea\\flinkUser\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")
          .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)
          })
          //*1000 要看是不是秒 秒*1000
          .assignAscendingTimestamps(_.timestamp*1000L)
    
        //3.transform 处理数据
        val processedStream = dataStream
          .filter(_.behavior == "pv")
          .keyBy(_.itemId)
          .timeWindow(Time.hours(1),Time.minutes(5))
          .aggregate(new CountAgg(),new WindowResult())
          .keyBy(_.windowEnd)
          .process(new TopNHotItems(3))
    
    
    
        //4.sink控制台输出
        processedStream.print()
    
        env.execute("hot items job")
      }
    
    }
    //自定义预聚合函数
    class CountAgg() extends AggregateFunction[UserBehavior,Long,Long]{ 
      override def createAccumulator(): Long = 0L
    
      override def add(in: UserBehavior, acc: Long): Long = acc + 1
      
      override def getResult(acc: Long): Long = acc
    
      override def merge(acc: Long, acc1: Long): Long = acc + acc1
    }
    //扩展 ---自定义预聚合函数计算平均数
    class AverageAgg() extends AggregateFunction[UserBehavior,(Long,Int),Double]{ 
      override def createAccumulator(): (Long, Int) = (0L,0)
    
      override def add(in: UserBehavior, acc: (Long, Int)): (Long, Int) = (acc._1 + in.timestamp,acc._2 + 1)
    
      override def getResult(acc: (Long, Int)): Double = acc._1/acc._2
    
      override def merge(acc: (Long, Int), acc1: (Long, Int)): (Long, Int) = (acc._1+acc1._1 ,acc._2+acc1._2)
    }
    //自定义窗口函数，输出ItemViewCount 第一个Long是预聚合最终输出的long 第三个long是ItemId
    class WindowResult() extends WindowFunction[Long,ItemViewCount,Long,TimeWindow]{ 
      override def apply(key: Long, window: TimeWindow, input: Iterable[Long],
                         out: Collector[ItemViewCount]): Unit = { 
        out.collect(ItemViewCount(key,window.getEnd,input.iterator.next()))
      }
    }
    //自定义的处理函数 第一个是windowEnd所以是long
    class TopNHotItems(topsize:Int) extends KeyedProcessFunction[Long,ItemViewCount,String]{ 
    
      private var itemState:ListState[ItemViewCount] = _
    
      override def open(parameters: Configuration): Unit = { 
        itemState = getRuntimeContext.getListState(new ListStateDescriptor[ItemViewCount]("item-state",
          classOf[ItemViewCount]))
      }
      override def processElement(i: ItemViewCount, context: KeyedProcessFunction
        [Long, ItemViewCount, String]#Context, collector: Collector[String]): Unit = { 
        //把每条数据存入状态列表
        itemState.add(i)
        //注册一个定时器 +1 是延迟时间
        context.timerService().registerEventTimeTimer(i.windowEnd + 1)
      }
    
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]
        #OnTimerContext, out: Collector[String]): Unit = { 
        //将所有state中的数据取出，放入到一个list buffer中
        val allItems:ListBuffer[ItemViewCount] = new ListBuffer()
        //将所有state中的数据取出，放入到一个list buffer中
        import scala.collection.JavaConversions._
        for (item <- itemState.get()){ 
          allItems += item
        }
        //按照Count大小排序 并取前N个 sortBy是升序
        val sortedItems = allItems.sortBy(_.count)(Ordering.Long.reverse).take(topsize)
    
        //清空状态 如果不想要下面的格式可以out.collect(sortedItems.toString())输出
        //UserBehavior(960222,4545844,2892802,pv,1511679926)
        itemState.clear()
        out.collect(sortedItems.toString())
    
        //将排名结果格式化输出 -1是之前注册定时器+1
        val result:StringBuilder = new StringBuilder()
        result.append("时间：").append(new Timestamp(timestamp-1)).append("\n")
        //输出每一个商品信息
        for (i <- sortedItems.indices){ 
          val currentItem = sortedItems(i)
          result.append("No").append(i+1).append(":")
            .append("商品ID =").append(currentItem.itemId)
            .append("浏览量").append(currentItem.count)
            .append("\n")
        }
        result.append("=====================")
        //控制输出频率
        Thread.sleep(1000)
    
        out.collect(result.toString())
    
      }
    
    }

结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020070521060256.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200705215113140.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70