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

逃离我推掉我的手 2023-02-24 15:24 41阅读 0赞

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

import java.sql.Timestamp
    
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.scala.{ OutputTag, StreamExecutionEnvironment}
    import org.apache.flink.api.scala._
    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
    
    
    
    //输入广告点击事件样例类
    case class AdClickEvent(userId:Long,adId:Long,province:String,city:String,timestamp:Long)
    //按照省份统计的输出结果样例类
    case class CountByProvince(windowEnd:String,provin:String,count:Long)
    //输出的黑名单报警信息
    case class BlackListWarning(userId:Long,adId:Long,msg:String)
    
    object AdStatisticsByGeo { 
    
      val blackListOutputTag = new OutputTag[BlackListWarning]("blacklist")
    
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        val resource = getClass.getResource("/AdClickLog.csv")
        val adEventStream = env.readTextFile(resource.getPath)
          .map(data=>{ 
            val dataArray = data.split(",")
            AdClickEvent(dataArray(0).trim.toLong,dataArray(1).trim.toLong,dataArray(2).trim,dataArray(3).trim,dataArray
            (4).trim.toLong)
          })
          .assignAscendingTimestamps(_.timestamp * 1000L)
    
        //自定义process function，过滤大量刷点击量的行为
        val filterBlackListStream = adEventStream
          .keyBy(data=>(data.userId,data.adId))
          .process(new FilterBlackListUser(100))
    
    
    
    
        //根据省份做分组，开窗聚合
        val adCountStream = filterBlackListStream
          .keyBy(_.province)
          .timeWindow(Time.hours(1),Time.seconds(5))
          .aggregate(new AdCountAgg(),new AdCountResult())
    
    
        adCountStream.print("count")
        filterBlackListStream.getSideOutput(blackListOutputTag).print("blacklist")
    
        env.execute("ad statistics job")
      }
    
    }
    class FilterBlackListUser(maxCount: Int) extends KeyedProcessFunction[(Long,Long),AdClickEvent,AdClickEvent]{ 
      //定义状态，保存当前用户对当前广告的点击量
      lazy val countState:ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count-state"
      ,classOf[Long]))
      //保存是否发送过黑名单的状态
      lazy val isSendBlackList:ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]
      ("issent-state",classOf[Boolean]))
      //保存定时器触发的时间戳
      lazy val resetTimer:ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("resettime-state"
      ,classOf[Long]))
      lazy val blackListOutputTag = new OutputTag[BlackListWarning]("blacklist")
    
    
      override def processElement(i: AdClickEvent, context: KeyedProcessFunction
        [(Long, Long), AdClickEvent, AdClickEvent]#Context, collector: Collector[AdClickEvent]): Unit = { 
        //取出count状态
        val curCount = countState.value()
    
        //如果是第一次处理，注册定时器,每天00:00触发
        if(curCount == 0){ 
          //现在是毫秒 /1000 变成秒 *60 分钟 *60 时 *24 天 =天数 +1是明天
          val ts = (context.timerService().currentProcessingTime()/(1000*60*60*24) + 1)*(1000*60*60*24)
          resetTimer.update(ts)
          context.timerService().registerProcessingTimeTimer(ts)
        }
    
        //判断计数是否达到上限，如果达到则加入黑名单
        if (curCount>=maxCount){ 
          //判断是否发送过黑明单，只发送一次
          if (!isSendBlackList.value()){ 
            isSendBlackList.update(true)
            //输出到测输出流
            context.output(blackListOutputTag,BlackListWarning(i.userId,i.adId,"Click over"+maxCount+"time today"))
          }
          return
        }
        //技术状态+1 输出数据到主流 意思就是curCount<maxCount 输出计数(不进入黑名单)
        countState.update(curCount+1)
        collector.collect(i)
    
      }
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]
        #OnTimerContext, out: Collector[AdClickEvent]): Unit = { 
        //定时器触发时，清空状态
        if (timestamp == resetTimer.value()){ 
          isSendBlackList.clear()
          countState.clear()
          resetTimer.clear()
        }
      }
    }
    
    
    //自定义预聚合函数
    class AdCountAgg() extends AggregateFunction[AdClickEvent,Long,Long]{ 
      override def createAccumulator(): Long = 0L
    
      override def add(in: AdClickEvent, acc: Long): Long = acc + 1
    
      override def getResult(acc: Long): Long = acc
    
      override def merge(acc: Long, acc1: Long): Long = acc + acc1
    }
    //自定义窗口函数处理 第一个试 预聚合函数输出的long 第三个key是keyby中的province
    class  AdCountResult() extends WindowFunction[Long,CountByProvince,String,TimeWindow]{ 
      override def apply(key: String, window: TimeWindow, input: Iterable[Long],
                         out: Collector[CountByProvince]): Unit = { 
        //如果你想要时间戳是String类型的new Timestamp(window.getEnd).toString
        out.collect(CountByProvince((new Timestamp(window.getEnd)).toString,key,input.iterator.next()))
      }
    }

![在这里插入图片描述][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]: /images/20230209/c041307ecc834ed093c647e61f243978.png