Flink-电商用户行为分析(页面广告点击量统计-黑名单过滤）_2

雨点打透心脏的1/2处 2022-11-21 04:10 210阅读 0赞

数据
链接：https://pan.baidu.com/s/1jX852lIDQ5Jm6nPRGe6pAA
提取码：km1w
对于广告的统计，最简单也最重要的就是页面广告的点击量，网站往往需要根据广告点击量来制定定价策略和调整推广方式，而且也可以借此收集用户的偏好信息。更加具体的应是，我们可以根据用户的地理位置进行划分，从而总结出不同省份用户对不同广告的偏好，这样更有助于广告的精准投放。
需求：
接下来我们就进行页面广告按照省份划分的点击量的统计，然后开一小时的时间窗口，
滑动距离为 5 秒，统计窗口内的点击事件数量。我们对一段时间内（比如一天内）的用户点击行为进行约束，如果对同一个广告点击超过一定限额（比如 100 次），应该把该户加入黑名单并报警，此后其点击行为不应该再统计。

广告点击量统计

我们使用两种方式来实现预聚合函数process和aggregate
process

  1. import java.sql.Timestamp
  3. import org.apache.flink.streaming.api.TimeCharacteristic
  4. import org.apache.flink.streaming.api.scala._
  5. import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
  6. import org.apache.flink.streaming.api.windowing.time.Time
  7. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  8. import org.apache.flink.util.Collector
  11. //定义输入输出样例类
  12. case class AdClickEvent(userId:Long,adId:Long,province:String,city:String,timestamp:Long)
  13. case class AdCountByProvince(province:String,windowEnd:String,count:Long)
  15. object AdAnalysisByProvince { 
  16.   def main(args: Array[String]): Unit = { 
  17.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  19.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  21.     val resource = getClass.getResource("/AdClickLog.csv")
  22.     val adLogStream = env.readTextFile(resource.getPath)
  23.       .map(data=> { 
  24.         val dataArray = data.split(",")
  25.         AdClickEvent(dataArray(0).toLong,dataArray(1).toLong,dataArray(2),dataArray(3),dataArray(4).toLong)
  26.       })
  27.       .assignAscendingTimestamps(_.timestamp * 1000L)
  29.     //按照province分组开窗聚合统计
  30.     val adCountStream = adLogStream
  31.       .keyBy(_.province)
  32.       .timeWindow(Time.hours(1),Time.seconds(5))
  33.       .process(new AdCount())
  35.     adCountStream.print()
  36.     env.execute("ad analysis job")
  37.   }
  39. }
  40. class AdCount() extends ProcessWindowFunction[AdClickEvent,AdCountByProvince,String,TimeWindow]{ 
  41.   override def process(key: String, context: Context, elements: Iterable[AdClickEvent],
  42.                        out: Collector[AdCountByProvince]): Unit = { 
  43.     val province = key
  44.     val windowEnd = new Timestamp(context.window.getEnd).toString
  45.     val count = elements.size
  46.     out.collect(AdCountByProvince(province,windowEnd,count))
  48.   }
  49. }

aggregate

  1. import java.sql.Timestamp
  3. import org.apache.flink.api.common.functions.AggregateFunction
  4. import org.apache.flink.streaming.api.TimeCharacteristic
  5. import org.apache.flink.streaming.api.scala._
  6. import org.apache.flink.streaming.api.scala.function.{ ProcessWindowFunction, WindowFunction}
  7. import org.apache.flink.streaming.api.windowing.time.Time
  8. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  9. import org.apache.flink.util.Collector
  12. //定义输入输出样例类
  13. case class AdClickEvent(userId:Long,adId:Long,province:String,city:String,timestamp:Long)
  14. case class AdCountByProvince(province:String,windowEnd:String,count:Long)
  16. object AdAnalysisByProvince { 
  17.   def main(args: Array[String]): Unit = { 
  18.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  19.     env.setParallelism(1)
  20.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  22.     val resource = getClass.getResource("/AdClickLog.csv")
  23.     val adLogStream = env.readTextFile(resource.getPath)
  24.       .map(data=> { 
  25.         val dataArray = data.split(",")
  26.         AdClickEvent(dataArray(0).toLong,dataArray(1).toLong,dataArray(2),dataArray(3),dataArray(4).toLong)
  27.       })
  28.       .assignAscendingTimestamps(_.timestamp * 1000L)
  30.     //按照province分组开窗聚合统计
  31.     val adCountStream = adLogStream
  32.       .keyBy(_.province)
  33.       .timeWindow(Time.hours(1),Time.seconds(5))
  34.       .aggregate(new AdCountAgg(),new AdCountResult())
  36.     adCountStream.print()
  37.     env.execute("ad analysis job")
  38.   }
  40. }
  41. //自定义预聚合函数，第一个In就是Map
  42. class AdCountAgg() extends AggregateFunction[AdClickEvent,Long,Long]{ 
  43.   override def createAccumulator(): Long = 0L
  45.   override def add(value: AdClickEvent, accumulator: Long): Long = accumulator + 1
  47.   override def getResult(accumulator: Long): Long = accumulator
  49.   override def merge(a: Long, b: Long): Long = a + b
  50. }
  51. //自定义窗口函数,第一个参数就预聚合函数最后输出的值，Long
  52. class AdCountResult() extends WindowFunction[Long,AdCountByProvince,String,TimeWindow]{ 
  53.   override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdCountByProvince]): Unit
  54.   = { 
  55.     out.collect(AdCountByProvince(key,new Timestamp(window.getEnd).toString,input.head))
  56.   }
  57. }

在这里插入图片描述

黑名单过滤

  1. import java.sql.Timestamp
  3. import org.apache.flink.api.common.functions.AggregateFunction
  4. import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
  5. import org.apache.flink.streaming.api.TimeCharacteristic
  6. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  7. import org.apache.flink.streaming.api.scala._
  8. import org.apache.flink.streaming.api.scala.function.{ ProcessWindowFunction, WindowFunction}
  9. import org.apache.flink.streaming.api.windowing.time.Time
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  11. import org.apache.flink.util.Collector
  14. //定义输入输出样例类
  15. case class AdClickEvent(userId:Long,adId:Long,province:String,city:String,timestamp:Long)
  16. case class AdCountByProvince(province:String,windowEnd:String,count:Long)
  17. //定义测输出流报警信息样例类
  18. case class BlackListWarning(userId:Long,adId:Long,msg:String)
  20. object AdAnalysisByProvinceBlack { 
  21.   def main(args: Array[String]): Unit = { 
  22.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  23.     env.setParallelism(1)
  24.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  26.     val resource = getClass.getResource("/AdClickLog.csv")
  27.     val adLogStream = env.readTextFile(resource.getPath)
  28.       .map(data=> { 
  29.         val dataArray = data.split(",")
  30.         AdClickEvent(dataArray(0).toLong,dataArray(1).toLong,dataArray(2),dataArray(3),dataArray(4).toLong)
  31.       })
  32.       .assignAscendingTimestamps(_.timestamp * 1000L)
  34.     //定义刷单行为过滤操作
  35.     val filterBlackListStream = adLogStream
  36.       .keyBy(data=> (data.userId,data.adId))  //按照用户和广告id进行集合
  37.       .process(new FilterBlackList(100L))
  40.     //按照province分组开窗聚合统计
  41.     val adCountStream = filterBlackListStream
  42.       .keyBy(_.province)
  43.       .timeWindow(Time.hours(1),Time.seconds(5))
  44.       .aggregate(new AdCountAgg(),new AdCountResult())
  47.     adCountStream.print()
  48.     filterBlackListStream.getSideOutput(new OutputTag[BlackListWarning]("blacklist")).print("blacklist")
  49.     env.execute("ad analysis job")
  50.   }
  52. }
  53. //实现自定义的ProcessFunction
  54. class FilterBlackList(maxClickCount: Long) extends KeyedProcessFunction[(Long,Long),AdClickEvent,AdClickEvent]{ 
  55.   //定义状态，需要保存当前用户对当前广告的点击量count
  56.   lazy val countState:ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count",classOf[Long]))
  57.   //标识位，用来表示用户是否已经在黑名单中
  58.   lazy val isSendState:ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-sent",
  59.     classOf[Boolean]))
  60.   override def processElement(value: AdClickEvent, ctx: KeyedProcessFunction[(Long,Long), AdClickEvent, AdClickEvent]
  61.     #Context, out: Collector[AdClickEvent]): Unit = { 
  62.     //取出状态数据
  63.     val curCount = countState.value()
  65.     //如果是第一个数据，那么注册第二天0点的定时器，用于清空状态
  66.     if (curCount == 0){ 
  67.       val ts = (ctx.timerService().currentProcessingTime() / (1000*60*60*24) + 1) * (1000*60*60*24)
  68.       ctx.timerService().registerProcessingTimeTimer(ts)
  69.     }
  70.     //判断count值是否达到上线，如果达到，并且之前没有输出过报警信息，那么报警
  71.     if (curCount >= maxClickCount){ 
  72.       if (!isSendState.value()){ 
  73.         ctx.output(new OutputTag[BlackListWarning]("blacklist"),BlackListWarning(value.userId,value.adId,
  74.         "click over"+maxClickCount+"times today"))
  75.         isSendState.update(true)
  76.       }
  77.       return
  78.     }
  80.     //count值加1
  81.     countState.update(curCount + 1)
  82.     out.collect(value)
  83.   }
  86.   //0点触发定时器，直接清空状态
  87.   override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]
  88.     #OnTimerContext, out: Collector[AdClickEvent]): Unit = { 
  89.     countState.clear()
  90.     isSendState.clear()
  91.   }
  92. }
  93. //自定义预聚合函数，第一个In就是Map
  94. class AdCountAgg() extends AggregateFunction[AdClickEvent,Long,Long]{ 
  95.   override def createAccumulator(): Long = 0L
  97.   override def add(value: AdClickEvent, accumulator: Long): Long = accumulator + 1
  99.   override def getResult(accumulator: Long): Long = accumulator
  101.   override def merge(a: Long, b: Long): Long = a + b
  102. }
  103. //自定义窗口函数,第一个参数就预聚合函数最后输出的值，Long
  104. class AdCountResult() extends WindowFunction[Long,AdCountByProvince,String,TimeWindow]{ 
  105.   override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdCountByProvince]): Unit
  106.   = { 
  107.     out.collect(AdCountByProvince(key,new Timestamp(window.getEnd).toString,input.head))
  108.   }
  109. }

在这里插入图片描述
我犯了一个低级错误，导致黑名单输出不出来
把这个

  1. val ts = (ctx.timerService().currentProcessingTime() / (1000*60*60*24) + 1) * (1000*60*60*24)

写成了

  1. val  ts = (ctx.timerService().currentProcessingTime() / ((1000*60*60*24) + 1) *  (1000*60*60*24))

发表评论

表情：
评论列表 （有 0 条评论，210人围观）

还没有评论，来说两句吧...

相关阅读