批处理：

  • 批处理主要操作大容量静态数据集，并在计算过程中完成后返回结果。可以认为，处理的是一个固定时间间隔分组的数据点集合。批处理模式中使用的数据集通常符号下列特征：
    • 有界：批处理数据集代表数据的有限集合
    • 持久：数据通常始终存储在某种类型的持久存储位置中
    • 大量：批处理操作通常是处理极为海量数据集的唯一方法

流处理：

  • 流处理可以对随时进入系统的数据进行计算。流处理方式无需针对整个数据集执行操作，而是对通过系统传输的每个数据执行操作。流处理中的数据集是”无边界的”，这就产生了几个重要的影响：
    • 可以处理几乎无限量的数据，但是同一时间只能处理一条数据，不同记录间只维持最少量的状态
    • 处理工作是基于事件的，除非明确停止是否没有”尽头”
    • 处理结果立刻可用，并会随着新数据的抵达继续更新

一、电商用户行为分析

  • 用户
    • 登陆方式
    • 上线时间点和时长
    • 页面停留和跳转
  • 商品
    • 收藏、评分、评价、打标签
    • 点击、浏览、购买、支付

由此可以从如下几个角度进行分析：

  • 统计分析
    • 点击
    • 热门商品、近期热门商品、分类热门商品、流量统计
  • 偏好统计
    • 收藏、喜欢、评分、打标签
    • 用户画像、推进列标（结合特征工程和机器学习算法）
  • 风险控制
    • 下订单、支付、登录
    • 刷单监控、订单失效监控、恶意登陆监控

二、项目模块设计

  • 实时统计分析
    • 实时热门商品统计
    • 实时热门页面流量统计
    • 实时访问流量统计
    • APP市场推广统计
    • 页面广告点击量统计
  • 业务流程及风险控制
    • 页面广告黑名单过滤
    • 恶意登陆监控
    • 订单支付失效监控
    • 支付实时对账

三、数据源

  • 用户行为数据
    • userBehavior.csv（userid，itemId，categoryid，behavior，timestamp）（加密后用户id，加密后商品id，商品所属类别id，用户行为类型，行为发生时间戳）
  • web服务器日志
    • apache.log（ip，userid，eventTime，method，url）（访问ip，访问userid，访问时间，访问方法，访问url）

四、需求

1、热门实时商品统计

  • 基本需求
    • 统计最近1小时的热门商品，每5分钟更新一次
    • 热门度用浏览次数(“pv”)来衡量
  • 解决思路
    • 在所有用户行为数据中，过滤出浏览(“pv”)行为进行统计
    • 在构建滑动窗口，窗口长度为1小时，滑动距离为5分钟

2、实时流量统计-热门页面

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

3、实时流量统计-PV和UV

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

4、市场营销分析-APP市场推广统计

  • 基本需求
    • 从埋点日志中，统计APP市场推广的数据指标
    • 按照不同的推广渠道，分别统计数据
  • 解决思路
    • 通过过滤日志中的用户行为，按照不同的渠道进行统计
    • 可以用processing function处理，得到自定义的输出数据信息

5、市场营销分析-页面广告统计

  • 基本需求
    • 从埋点日志中，统计每小时页面广告的点击量，5秒刷新一次，并按照不同省份进行划分
    • 对于“刷单”式的频繁点击行为进行过滤，并将该用户加入黑名单
  • 解决思路
    • 根据省份进行分组，创建长度为1小时、滑动距离为5秒的时间窗口进行统计
    • 可以用processfunction进行黑名单过滤，检测用户对同一广告的点击量，如果超过上限则将该用户信息以侧输出流输出到黑名单

6、恶意登陆监控

  • 基本需求
    • 用户在短时间内频繁登陆失败，有程序恶意攻击的可能
    • 同一用户（可以式不同ip）在2秒内连续登陆失败，需求报警
  • 解决思路
    • 将用户登陆失败行为存入ListState，设定定时器2秒后触发，查看ListState中有几次失败登陆
    • 更加精确的检测，可以使用CEP库实现事件流的模式匹配

7、订单支付实时监控

  • 基本需求
    • 用户下单后，应设置订单失效时间，以提高用户支付的意愿，并降低系统风险
    • 用户下单后15分钟未支付，则输出监控信息
  • 解决思路
    • 利用CEP库进行事件流的模式匹配，并设定匹配的时间间隔
    • 也可以利用状态编程，用processfunction实现处理逻辑

8、订单支付实时对账

  • 基本需求
    • 用户下单并支付后，应查询到账信息，进行实时对账
    • 如果有不匹配的支付信息或者到账信息，输出提示信息
  • 解决思路
    • 从两条流中分别读取订单支付信息和到账信息，合并处理
    • 用connect连接合并两条流，用coProcessFunction做匹配处理

五、实现代码

5.2、实时流量统计-热门页面

  1. package com.learn.networkflow
  3. import java.net.URL
  4. import java.sql.Timestamp
  5. import java.text.SimpleDateFormat
  6. import java.util.Map
  7. import java.{lang, util}
  9. import org.apache.flink.api.common.functions.AggregateFunction
  10. import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor}
  11. import org.apache.flink.streaming.api.TimeCharacteristic
  12. import org.apache.flink.streaming.api.datastream.{DataStreamSource, KeyedStream, SingleOutputStreamOperator, WindowedStream}
  13. import org.apache.flink.streaming.api.environment._
  14. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  15. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  16. import org.apache.flink.streaming.api.functions.windowing.WindowFunction
  17. import org.apache.flink.streaming.api.scala.OutputTag
  18. import org.apache.flink.api.scala._
  19. import org.apache.flink.streaming.api.windowing.time.Time
  20. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  21. import org.apache.flink.util.Collector
  23. import scala.collection.mutable.ListBuffer
  26. //定义输入数据
  27. //访问ip，访问userid，访问时间，访问方法，访问url
  28. case class ApacheLogEvent(ip:String , userid: String , eventTime: Long, method:String ,url:String)
  29. //窗口聚合结果样例类
  30. case class PageViewCount(url:String , windowEnd: Long , count: Long)
  33. /**
  34.   * ## 实时流量统计-热门页面
  35.   *
  36.   * - 基本需求
  37.   *   - 从web服务器日志中，统计实时的热门访问页面
  38.   *   - 统计每分钟的ip访问量，取出访问量最大的5个地址，每5秒更新一次
  39.   * - 解决思路
  40.   *   - 将服务器日志中的事件，转换成时间戳，作为EventTime
  41.   *   - 构建滑动窗口，窗口长度为1分钟，滑动距离为5秒
  42.   *
  43.   */
  44. object HotPageNetworkFlow {
  46.   def main(args: Array[String]): Unit = {
  48.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  49.     env.setParallelism(1)
  50.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  52. //    val resource: URL = getClass.getResource("/apache.log")
  53. //    val inputStream: DataStreamSource[String] = env.readTextFile(resource.getPath)
  55.     val inputStream: DataStreamSource[String] = env.socketTextStream("localhost",7777)
  57.     val dataStream: SingleOutputStreamOperator[ApacheLogEvent] = inputStream.map(data => {
  58.       val arr: Array[String] = data.split(" ")
  59.       val format = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
  60.       val ts: Long = format.parse(arr(2)).getTime
  61.       ApacheLogEvent(arr(0), arr(1), ts, arr(3), arr(4))
  62.     }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(1)) {
  63.       override def extractTimestamp(element: ApacheLogEvent): Long = element.eventTime
  64.     })
  66.     dataStream.print("data")
  68.     val filterStream: SingleOutputStreamOperator[ApacheLogEvent] = dataStream.filter(_.method=="GET")
  70.     val keyedByStream: KeyedStream[ApacheLogEvent, String] = filterStream.keyBy(_.url)
  72. //    val windowStream: WindowedStream[ApacheLogEvent, String, TimeWindow] = keyedByStream.timeWindow(Time.minutes(10) , Time.seconds(5))
  74. //    implicit val typeInfo = TypeInformation.of(classOf[ApacheLogEvent])//隐式转换
  75.     val tag = new OutputTag[ApacheLogEvent]("late")
  77.     val aggStream = dataStream
  78.       .filter(_.method == "GET")
  79.       .keyBy(_.url)
  80.       .timeWindow(Time.minutes(10), Time.seconds(5))
  81.       .allowedLateness(Time.minutes(1))//延迟窗口多1分钟再关闭
  82.       .sideOutputLateData(tag) //标记迟到数据（即窗口关闭后的数据）进入侧输出流，迟到数据最晚时间 = watermark - windowside - lateness
  83.       .aggregate(new PageCountAgg(), new PageViewCountWindowResult)
  85.     aggStream.print("agg")
  87.     val result: SingleOutputStreamOperator[String] = aggStream.keyBy(_.windowEnd)
  88.       .process(new TopNHotPages(5))
  89.     result.print("result")
  91.     aggStream.getSideOutput(tag).print("late")//延迟数据输出
  93.     env.execute("Hot Pages Job")
  95.   }
  97. }
  99. class PageCountAgg extends AggregateFunction[ApacheLogEvent , Long , Long]{
  100.   override def createAccumulator(): Long = 0L
  102.   override def add(value: ApacheLogEvent, accumulator: Long): Long = accumulator+1L
  104.   override def getResult(accumulator: Long): Long = accumulator
  106.   override def merge(a: Long, b: Long): Long = a+b
  107. }
  109. class PageViewCountWindowResult extends WindowFunction[Long , PageViewCount , String , TimeWindow]{
  110.   override def apply(key: String, window: TimeWindow, input: lang.Iterable[Long], out: Collector[PageViewCount]): Unit = {
  111.     out.collect(PageViewCount(key , window.getEnd , input.iterator().next()))
  112.   }
  113. }
  115. class TopNHotPages(n: Int) extends KeyedProcessFunction[Long , PageViewCount , String]{
  117.   //保存 url-count
  118.   lazy val mapState: MapState[String,Long] = getRuntimeContext.getMapState(new MapStateDescriptor[String,Long]("pageViewCount",classOf[String],classOf[Long]))
  120.   override def processElement(value: PageViewCount, ctx: KeyedProcessFunction[Long, PageViewCount, String]#Context, out: Collector[String]): Unit = {
  122.     mapState.put(value.url , value.count)
  123.     println("当前windownd = "+value.windowEnd)
  124.     ctx.timerService().registerEventTimeTimer(value.windowEnd+1)
  125.     ctx.timerService().registerEventTimeTimer(value.windowEnd+60000) //注册一分钟后触发当窗口关闭，不再输出结果时，进行清空状态,和延迟关闭窗口时间长度一致
  126.   }
  128.   override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, PageViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
  130.     val iter: util.Iterator[Map.Entry[String, Long]] = mapState.entries().iterator()
  131.     val allPageViewCount:ListBuffer[(String , Long)] = ListBuffer()
  133.     while (iter.hasNext){
  134.       val entry = iter.next()
  135.       allPageViewCount+= ((entry.getKey , entry.getValue))
  136.     }
  138.     //清空状态
  139.     if (timestamp == ctx.getCurrentKey + 60000){
  140.       mapState.clear()
  141.       return
  142.     }
  144.     //排序，取前几的数据
  145.     val sortPageViewCounts: ListBuffer[(String, Long)] = allPageViewCount.sortWith(_._2 > _._2).take(n)
  147.     val res:StringBuilder = new StringBuilder
  148.     res.append("窗口结束时间：").append(new Timestamp(timestamp-1)).append("\n")
  149.     for (i <- sortPageViewCounts.indices){
  150.       val item: (String , Long) = sortPageViewCounts(i)
  151.       res.append("No").append(i+1).append(": ")
  152.         .append("url = ").append(item._1).append("\t")
  153.         .append("热门度 = ").append(item._2).append("\n")
  154.     }
  156.     res.append("==============================\n\n")
  157.     //    Thread.sleep(1000)
  158.     out.collect(res.toString)
  159.   }
  160. }

5.3、实时流量统计-PV和UV

  1. package com.learn.networkflow
  3. import java.net.URL
  5. import org.apache.flink.api.common.functions.{AggregateFunction, MapFunction}
  6. import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
  7. import org.apache.flink.streaming.api.TimeCharacteristic
  8. import org.apache.flink.streaming.api.datastream.{DataStreamSource, SingleOutputStreamOperator}
  9. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  10. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  11. import org.apache.flink.streaming.api.scala._
  12. import org.apache.flink.streaming.api.scala.function.WindowFunction
  13. import org.apache.flink.streaming.api.windowing.time.Time
  14. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  15. import org.apache.flink.util.Collector
  17. import scala.util.Random
  19. //定义输出数据样例类
  20. case class UserBehavior(userid:Long , itemId:Long , categoryid:Int , behavior:String , timestamp:Long)
  21. //加密后用户id，加密后商品id，商品所属类别id，用户行为类型，行为发生时间戳
  23. //定义窗口聚合结果样例类
  24. case class PvCount( windowEnd: Long , count:Long)
  26. /**
  27.   * 实时流量统计-PV和UV
  28.   *
  29.   * - 基本需求
  30.   *   - 从埋点日志中，统计实时的PV和UV
  31.   *   - 统计每小时的访问量（PV），并对用户进行去重
  32.   * - 解决思路
  33.   *   - 统计埋点日志中的pv行为，利用set数据结构进行去重
  34.   *   - 对于超大规模的数据，可以考虑用布隆过滤器进行去重
  35.   */
  36. object PageView {
  38.   def main(args: Array[String]): Unit = {
  40.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  41.     env.setParallelism(1)
  42.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  44. //    val resource: URL = getClass.getResource("/apache.log")
  45. //    val inputStream: DataStream[String] = env.readTextFile(resource.getPath)
  46.     val inputStream: DataStream[String] = env.socketTextStream("localhost",7777)
  49.     val dataStream: DataStream[UserBehavior] = inputStream.map(
  50.       data => {
  51.         val arr: Array[String] = data.split(",")
  52.         UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3).toString, arr(4).toLong)
  53.       }
  54.     ).assignAscendingTimestamps(_.timestamp*1000)
  56.     val pvStream = dataStream.filter(_.behavior == "pv")
  57. //      .map(data=>("pv",1L))
  58.       .map(new MyKeyByFunction)
  59.       .keyBy(_._1)   //由于key都是一样的，所有数据都分到同一个slot，这样就会导致程序的并行度没有意义，可以使用自定义map组成一个乱序的key
  60.       .timeWindow(Time.seconds(5))
  61.       .aggregate(new PvCountAgg() , new PvCountWindowResult)
  63.     pvStream.print("pvStream") // 这里的聚合结果每来一个数据就会聚合一次，所以需要再增加一个window分组按照每个分组输出
  65.     val result = pvStream
  66.         .keyBy(_.windowEnd)  //再把每个window的结果聚合
  67.         .process(new TotalPvCountResult) //根据每个key进行分组，定时再window结束后再统一输出
  69.     result.print("result")
  71.     env.execute("pv Count Job")
  72.   }
  74. }
  76. class PvCountAgg() extends AggregateFunction[(String,Long) , Long , Long]{
  77.   override def createAccumulator(): Long = 0L
  79.   override def add(value: (String, Long), accumulator: Long): Long = accumulator+1
  81.   override def getResult(accumulator: Long): Long = accumulator
  83.   override def merge(a: Long, b: Long): Long = a+b
  84. }
  86. class PvCountWindowResult extends WindowFunction[Long , PvCount , String , TimeWindow]{
  87.   override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[PvCount]): Unit = {
  89.     out.collect(PvCount( window.getEnd , input.iterator.next()))
  91.   }
  92. }
  94. class MyKeyByFunction extends MapFunction[UserBehavior, (String,Long)]{
  95.   override def map(value: UserBehavior): (String, Long) = {
  96.     (new Random().nextString(10) , 1)
  97.   }
  98. }
  100. class TotalPvCountResult extends KeyedProcessFunction[Long , PvCount , PvCount]{
  102.   lazy val totalPvCountResultState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("total-pvCount",classOf[Long]))
  104.   override def processElement(value: PvCount, ctx: KeyedProcessFunction[Long, PvCount, PvCount]#Context, out: Collector[PvCount]): Unit = {
  106.     totalPvCountResultState.update(value.count + totalPvCountResultState.value())
  108.     ctx.timerService().registerEventTimeTimer(value.windowEnd+1)
  110.   }
  112.   override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, PvCount, PvCount]#OnTimerContext, out: Collector[PvCount]): Unit = {
  114.     out.collect(PvCount(ctx.getCurrentKey , totalPvCountResultState.value()))
  115.     totalPvCountResultState.clear()
  117.   }
  118. }

5.4、UV统计-全窗口

  1. package com.learn.networkflow
  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.AllWindowFunction
  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
  10. //定义窗口聚合结果样例类
  11. case class UvCount( windowEnd: Long , count:Long)
  13. /**
  14.   * uv统计
  15.   */
  16. object UvCount {
  18.   def main(args: Array[String]): Unit = {
  20.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  21.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  22.     env.setParallelism(1)
  24.     val inputStream: DataStream[String] = env.socketTextStream("localhost",7777)
  26.     val dataStream: DataStream[UserBehavior] = inputStream.map(
  27.       data => {
  28.         val arr: Array[String] = data.split(",")
  29.         UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3).toString, arr(4).toLong)
  30.       }
  31.     ).assignAscendingTimestamps(_.timestamp*1000)
  33.     val uvStream = dataStream.filter(_.behavior == "pv")
  34.       .timeWindowAll(Time.seconds(5))
  35.       .apply(new UvCountResult())
  37.   }
  39. }
  41. /**
  42.   * 通过set实现数据去重，但是在实际生产环境这种方法会造成内存过大造成资源浪费
  43.   * 可以实现通过布隆过滤器进行去重
  44.   */
  45. class UvCountResult() extends AllWindowFunction[UserBehavior , UvCount , TimeWindow]{
  46.   override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {
  48.    var userIdSet = Set[Long]()
  50.     val iter: Iterator[UserBehavior] = input.iterator
  52.     while (iter.hasNext){
  54.       val behavior: UserBehavior = iter.next()
  55.       userIdSet+=behavior.userid
  57.     }
  58.     out.collect(UvCount(window.getEnd , userIdSet.size))
  59.   }
  60. }

5.5、市场营销分析

5.5.1、APP市场推广统计

  1. package com.market.analysis
  3. import java.lang
  4. import java.sql.Timestamp
  6. import org.apache.flink.api.common.functions.AggregateFunction
  7. import org.apache.flink.streaming.api.TimeCharacteristic
  8. import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
  9. import org.apache.flink.streaming.api.scala._
  10. import org.apache.flink.streaming.api.scala.function.{ProcessWindowFunction, WindowFunction}
  11. import org.apache.flink.streaming.api.windowing.time.Time
  12. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  13. import org.apache.flink.util.Collector
  15. import scala.util.Random
  17. //输入数据类型
  18. case class MarketUserBehavior(userId: String , behavior: String , channel: String , timestamp: Long)
  19. //输出数据类型
  20. case class MarketViewCount(windowStart: String , windowEnd: String , channel: String , behavior: String , count: Long)
  22. //自定义数据测试源
  23. class SimulatedSource extends RichSourceFunction[MarketUserBehavior]{
  25.   //定义数据是否产生标志位
  26.   var running = true
  28.   val behaviorSet: Seq[String] = Seq("view" , "click" , "download" , "install" , "uninstall")
  29.   val channelSet: Seq[String] = Seq("appstore" , "weibo" , "wechat" , "tieba")
  30.   val random:Random = Random
  32.   //生产数据
  33.   override def run(ctx: SourceFunction.SourceContext[MarketUserBehavior]): Unit = {
  35.     val maxCount: Long = Long.MaxValue
  36.     var count = 0L
  38.     while (running && count < maxCount){
  39.       val userId: String = random.nextString(20)
  40.       val behavior: String = behaviorSet(random.nextInt(behaviorSet.size))
  41.       val channel: String = channelSet(random.nextInt(channelSet.size))
  42.       val timestamp: Long = System.currentTimeMillis()
  44.       ctx.collect(MarketUserBehavior(userId,behavior,channel,timestamp))
  45.       count+=1
  46.       Thread.sleep(500)
  47.     }
  48.   }
  49.   //关闭数据产生
  50.   override def cancel(): Unit = {
  51.     running = false
  52.   }
  53. }
  55. /**
  56.   * 统计分析根据不同渠道、不同行为 在过去1小时的数量
  57.   */
  58. object AppMarketByChannel {
  60.   def main(args: Array[String]): Unit = {
  61.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  62.     env.setParallelism(1)
  63.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  65.     val dataStream: DataStream[MarketUserBehavior] = env.addSource(new SimulatedSource)
  66.       .assignAscendingTimestamps(_.timestamp)
  68.     //统计
  69.     val filterStream: DataStream[MarketUserBehavior] = dataStream.filter(_.behavior != "uninstall")
  71.     val keyByStream: KeyedStream[MarketUserBehavior, (String, String)] = filterStream.keyBy(data => {
  72.       (data.channel, data.behavior) //可以传入一个二元组，表示根据两个字段进行keyBy
  73.     })
  75.     val windowStream: WindowedStream[MarketUserBehavior, (String, String), TimeWindow] = keyByStream.timeWindow(Time.days(1) , Time.seconds(5))
  77.     /**
  78.       * 实现一、通过Agg函数和WindowFunction实现
  79.       */
  80.     val result: DataStream[MarketViewCount] = windowStream.aggregate(new AppMarkAggregate , new AppMarkWindowCount)
  81.     result.print("res1")
  83.     /**
  84.       * 实现二、通过processFunction实现
  85.       */
  86.     val res2: DataStream[MarketViewCount] = windowStream.process(new MarketCountByChannel)
  87.     res2.print("res2")
  90.     env.execute("AppMarket Job")
  91.   }
  93. }
  95. /**
  96.   * 聚合
  97.   */
  98. class AppMarkAggregate extends AggregateFunction[MarketUserBehavior , Long , Long]{
  99.   override def createAccumulator(): Long = 0L
  101.   override def add(value: MarketUserBehavior, accumulator: Long): Long = accumulator+1
  103.   override def getResult(accumulator: Long): Long = accumulator
  105.   override def merge(a: Long, b: Long): Long = a+b
  106. }
  108. /**
  109.   * 窗口统计
  110.   */
  111. class AppMarkWindowCount extends WindowFunction[Long , MarketViewCount , (String , String) , TimeWindow]{
  112.   override def apply(key: (String, String), window: TimeWindow, input: Iterable[Long], out: Collector[MarketViewCount]): Unit = {
  113.     out.collect(MarketViewCount(new Timestamp(window.getStart).toString , new Timestamp(window.getEnd).toString , key._1 , key._2 , input.iterator.next()))
  114.   }
  115. }
  117. /**
  118.   * 1、前面提到的 ReduceFunction 和 AggregateFunction 都是基于中间状态实现增量计算的窗口函数，虽然已经满足绝大多数场景，
  119.   * 但在某些情况下，统计更复杂的指标可能需要依赖于窗口中所有的数据元素，或需要操作窗口中的状态数据和窗口元数据，
  120.   * 这时就需要使用到 ProcessWindowsFunction，ProcessWindowsFunction 能够更加灵活地支持基于窗口全部数 据元素的结果计算，
  121.   * 例如对整个窗口数据排序取 TopN，这样的需要就必须使用 ProcessWindowFunction。
  122.   * 
  123.   * 2、ProcessWindowFunction获得一个包含窗口所有元素的可迭代器，以及一个具有时间和状态信息访问权的上下文对象，这使得它比其他窗口函数提供更大的灵活性。
  124.   * 这是以性能和资源消耗为代价的，因为元素不能增量地聚合，而是需要在内部缓冲，直到认为窗口可以处理为止，使用该函数需要注意数据量，数据量太大，全量数据保存在内存中，会造成内存溢出。
  125.   *
  126.   */
  127. class MarketCountByChannel extends ProcessWindowFunction[MarketUserBehavior , MarketViewCount , (String,String) , TimeWindow]{
  128.   override def process(key: (String, String), context: Context, elements: Iterable[MarketUserBehavior], out: Collector[MarketViewCount]): Unit = {
  130.     val start: Long = context.window.getStart
  131.     val end: Long = context.window.getEnd
  132.     val channel = key._1
  133.     val behavior = key._2
  134.     val size: Int = elements.size
  135.     out.collect(MarketViewCount(new Timestamp(start).toString , new Timestamp(end).toString , key._1 , key._2 , size))
  136.   }
  137. }

5.5.2、页面广告分析

根据不同的省份地区，统计对不同广告的点击量，并且过滤点击超过100次的用户

  1. package com.market.analysis
  3. import java.sql.Timestamp
  4. import java.util.UUID
  6. import org.apache.flink.api.common.functions.AggregateFunction
  7. import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
  8. import org.apache.flink.streaming.api.TimeCharacteristic
  9. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  10. import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
  11. import org.apache.flink.streaming.api.scala._
  12. import org.apache.flink.streaming.api.scala.function.WindowFunction
  13. import org.apache.flink.streaming.api.windowing.time.Time
  14. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  15. import org.apache.flink.util.Collector
  17. import scala.util.Random
  19. //输入
  20. case class AdClickLog(userId:String , adId: String , province: String , city: String , ts: Long)
  22. //输出
  23. case class AdClickCountByProvince(windowEnd: String , province: String , count: Long)
  24. case class BlackListUserWarning(userId: String , adId: String , des: String)
  26. //自定义数据测试源
  27. class AdClickDataSource extends RichSourceFunction[AdClickLog]{
  29.   //定义数据是否产生标志位
  30.   var running = true
  32.   val gdSet: Seq[String] = Seq("广州" , "深圳" , "珠海" , "中山" , "顺德")
  33.   val beijingSet: Seq[String] = Seq("东城区" , "西城区" , "朝阳区" , "崇文区" , "宣武区")
  34.   val fujianSet: Seq[String] = Seq("福州" , "厦门" , "漳州" , "泉州" )
  35.   val shandongSet: Seq[String] = Seq("青岛" , "淄博" , "枣庄" , "烟台" , "济南" , "潍坊" , "济宁" , "日照")
  36.   val provinceSet: Seq[String] = Seq("广东" , "北京" , "福建" , "山东")
  37.   private val provincesMap: Map[String, Seq[String]] = Map("广东"->gdSet , "北京"->beijingSet , "福建"->fujianSet , "山东"->shandongSet)
  38.   val random:Random = Random
  40.   //生产数据
  41.   override def run(ctx: SourceFunction.SourceContext[AdClickLog]): Unit = {
  43.     val maxCount: Long = Long.MaxValue
  44.     var count = 0L
  46.     while (running && count < maxCount){
  47.       val userId: String = UUID.randomUUID().toString.replace("-","")
  48.       val adId: String = random.nextInt(100).toString
  49.       val province: String = provinceSet(random.nextInt(provinceSet.size))
  50.       val city: String = provincesMap.get(province).get(provincesMap.get(province).size)
  52.       val timestamp: Long = System.currentTimeMillis()
  54.       ctx.collect(AdClickLog(userId,adId,province,city,timestamp))
  55.       count+=1
  57.       Thread.sleep(50)
  58.     }
  59.   }
  61.   //关闭数据产生
  62.   override def cancel(): Unit = {
  63.     running = false
  64.   }
  65. }
  67. /**
  68.   * 主方法实现
  69.   */
  70. object AdClinkAnalysis {
  72.   def main(args: Array[String]): Unit = {
  74.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  75.     env.setParallelism(1)
  76.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  78.     val dataStream: DataStream[AdClickLog] = env.addSource(new AdClickDataSource)
  79.       .assignAscendingTimestamps(_.ts)
  81.     /**
  82.       * 过滤黑名单用户,对于同一用户每点击同一个广告计数加1，然后通过keyBy进行分组统计，超过一定数据量输出到侧输出流
  83.       */
  84.     val filterKeyByStream: KeyedStream[AdClickLog, (String, String)] = dataStream.keyBy(data=>{(data.userId,data.adId)})
  85.     val filterBlackUserStream: DataStream[AdClickLog] = filterKeyByStream.process(new FilterBlackListUserResult)
  87.     val keyByStream: KeyedStream[AdClickLog, String] = filterBlackUserStream.keyBy(_.province)
  89.     val windowStream: WindowedStream[AdClickLog, String, TimeWindow] = keyByStream.timeWindow(Time.days(1) , Time.seconds(5))
  91.     val result: DataStream[AdClickCountByProvince] = windowStream.aggregate(new AdClickAgg,new AdClickWindowRes)
  92.     result.print("res")
  94.     /**
  95.       *
  96.       */
  97.     filterBlackUserStream.getSideOutput(new OutputTag[BlackListUserWarning]("warning")).print("warning")
  99.     env.execute("AdClickJob")
  100.   }
  102. }
  104. class AdClickAgg extends AggregateFunction[AdClickLog , Long , Long]{
  105.   override def createAccumulator(): Long = 0L
  107.   override def add(value: AdClickLog, accumulator: Long): Long = accumulator+1
  109.   override def getResult(accumulator: Long): Long = accumulator
  111.   override def merge(a: Long, b: Long): Long = a+b
  112. }
  114. class AdClickWindowRes extends WindowFunction[Long,AdClickCountByProvince,String,TimeWindow]{
  115.   override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickCountByProvince]): Unit = {
  116.     out.collect(AdClickCountByProvince(new Timestamp(window.getEnd).toString , key , input.iterator.next()))
  117.   }
  118. }
  121. /**
  122.   * 黑名单过滤器
  123.   * 实现同一用户多次点击同一个广告即禁止其访问
  124.   */
  125. class FilterBlackListUserResult extends KeyedProcessFunction[(String,String) , AdClickLog , AdClickLog]{
  126.   //用户点击量
  127.   lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count",classOf[Long]))
  128.   //清空当日黑名单时间
  129.   lazy val resetTimerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("reset-ts",classOf[Long]))
  130.   //是否触发了黑名单
  131.   lazy val isBlackState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-Black",classOf[Boolean]))
  133.   lazy val maxcount = 10
  135.   override def processElement(value: AdClickLog, ctx: KeyedProcessFunction[(String, String), AdClickLog, AdClickLog]#Context, out: Collector[AdClickLog]): Unit = {
  137.     val curCount: Long = countState.value()
  139.     //1、判断只要是第一个数据来了，注册当天清空状态的定时器
  140.     if (curCount == 0){
  141.       val ts = (ctx.timerService().currentProcessingTime()/(1000*3600*24)+1)*(1000*3600*24) - 8*3600*1000 // 获取第二天0点的时间戳
  142.       resetTimerTsState.update(ts)
  143.       ctx.timerService().registerProcessingTimeTimer(ts)
  144.     }
  146.     //2、判断count值是否到达阈值，到达了则加入黑名单，输出到侧输出流
  147.     if (curCount >= maxcount){
  148.       if(!isBlackState.value()){
  149.         isBlackState.update(true)
  150.         ctx.output(new OutputTag[BlackListUserWarning]("warning"),BlackListUserWarning(value.userId , value.adId , "当前用户点击超过限制次数"))
  151.       }
  152.       return
  153.     }
  155.     //3、没有到达点击限制
  156.     countState.update(curCount+1)
  157.     out.collect(value)
  158.   }
  160.   override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(String, String), AdClickLog, AdClickLog]#OnTimerContext, out: Collector[AdClickLog]): Unit = {
  161.     if (timestamp == resetTimerTsState.value()){
  162.       isBlackState.clear()
  163.       resetTimerTsState.clear()
  164.       countState.clear()
  165.     }
  166.   }
  167. }

5.6、恶意登录监控

5.6.1、第一个版本

  1. package com.learn.loginfail
  3. import java.util
  5. import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
  6. import org.apache.flink.streaming.api.TimeCharacteristic
  7. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  8. import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
  9. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  10. import org.apache.flink.streaming.api.scala._
  11. import org.apache.flink.streaming.api.windowing.time.Time
  12. import org.apache.flink.util.Collector
  14. import scala.collection.mutable.ListBuffer
  15. import scala.util.Random
  16. import java.sql.Timestamp
  18. /**
  19.   * 输入数据,登陆数据
  20.   */
  21. case class LoginEventData(userId:String , ip:String , eventType:String , timestamp:Long)
  23. /**
  24.   * 输出样例类
  25.   */
  26. case class LoginFailWarning(userId:String , firstFailTime:Long , lastFailTime:Long , warningMsg:String)
  28. //自定义数据测试源
  29. class LoginEventDataSource extends RichSourceFunction[LoginEventData]{
  31.   //定义数据是否产生标志位
  32.   var running = true
  34.   val eventTypeSet: Seq[String] = Seq("success" , "fail" )
  35.   val random:Random = Random
  37.   //生产数据
  38.   override def run(ctx: SourceFunction.SourceContext[LoginEventData]): Unit = {
  39.     val maxCount: Long = Long.MaxValue
  40.     var count = 0L
  42.     while (running && count < maxCount){
  43.       val userId: String = random.nextInt(20).toString
  44.       val ip: String = random.nextInt(100000).toString
  45.       val eventType: String = eventTypeSet(random.nextInt(eventTypeSet.size))
  46.       val timestamp: Long = System.currentTimeMillis()
  48.       ctx.collect(LoginEventData(userId,ip,eventType,timestamp))
  49.       count+=1
  51.       Thread.sleep(500)
  52.     }
  53.   }
  55.   //关闭数据产生
  56.   override def cancel(): Unit = {
  57.     running = false
  58.   }
  59. }
  61. /**
  62.   * 检测同一用户在一段时间内，多次登陆失败则报警行为
  63.   */
  64. object LoginCheckAnalysis {
  66.   def main(args: Array[String]): Unit = {
  68.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  69.     env.setParallelism(1)
  70.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  72.     val inputStream: DataStream[LoginEventData] = env.addSource(new LoginEventDataSource)
  73.         .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEventData](Time.seconds(2)) {
  74.           override def extractTimestamp(element: LoginEventData): Long = element.timestamp
  75.         })
  77.     //进行判断和检测,如果两秒内登陆失败则输出报警信息
  78.     val loginFailWarningStream: DataStream[LoginFailWarning] = inputStream.keyBy(_.userId)
  79.       .process(new LoginFailWarningResult(2))
  81.     loginFailWarningStream.map(data=>{
  82.       (data.userId , new Timestamp(data.firstFailTime).toString , new Timestamp(data.lastFailTime).toString , data.warningMsg)
  83.     }).print("fail")
  85.     env.execute("LoginCheckJob")
  86.   }
  88. }
  90. //处理类
  91. class LoginFailWarningResult(failMaxSize: Int) extends KeyedProcessFunction[String , LoginEventData , LoginFailWarning]{
  93.   //保存所有登陆失败事件
  94.   lazy val loginFailListState: ListState[LoginEventData] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEventData]("failEvent-list",classOf[LoginEventData]))
  95.   //保存定时器时间戳
  96.   lazy val timeValueState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("time-ts",classOf[Long]))
  98.   //每来一条数据，对应调用一次processElement
  99.   override def processElement(value: LoginEventData, ctx: KeyedProcessFunction[String, LoginEventData, LoginFailWarning]#Context, out: Collector[LoginFailWarning]): Unit = {
  100. //    println("数据进入process = "+value)
  102.     //判断是否失败事件
  103.     if (value.eventType == "fail"){
  104.       loginFailListState.add(value)
  105.       //注册2秒后触发报警定时器
  106.       ctx.timerService().registerEventTimeTimer(value.timestamp+2000L)
  107.       timeValueState.update(value.timestamp+2000L)
  108.     }else{
  109.       //如果是成功，直接清空状态 和 定时器
  110.       loginFailListState.clear()
  111.       ctx.timerService().deleteEventTimeTimer(timeValueState.value())
  112.       timeValueState.clear()
  113.     }
  114.   }
  116.   override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, LoginEventData, LoginFailWarning]#OnTimerContext, out: Collector[LoginFailWarning]): Unit = {
  118.     val allLoginFailEventData = new ListBuffer[LoginEventData]
  120.     val iter: util.Iterator[LoginEventData] = loginFailListState.get().iterator()
  122.     while (iter.hasNext){
  123.       allLoginFailEventData += iter.next()
  124.     }
  126.     //当超过一定失败次数，触发输出
  127.     if (allLoginFailEventData.size >= failMaxSize){
  128.       out.collect(new LoginFailWarning(allLoginFailEventData.head.userId ,
  129.         allLoginFailEventData.head.timestamp, // 第一个元素
  130.         allLoginFailEventData.last.timestamp, // 最后一个元素
  131.         "login fail in 2 s for "+ allLoginFailEventData.length +" times"
  132.       ))
  133.     }
  134.     loginFailListState.clear()
  135.     timeValueState.clear()
  136.   }
  137. }

5.6.2、另一种版本

也不是最好的解决

  1. package com.learn.loginfail
  3. import java.util
  5. import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
  6. import org.apache.flink.streaming.api.TimeCharacteristic
  7. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  8. import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
  9. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  10. import org.apache.flink.streaming.api.scala._
  11. import org.apache.flink.streaming.api.windowing.time.Time
  12. import org.apache.flink.util.Collector
  14. import scala.collection.mutable.ListBuffer
  15. import scala.util.Random
  16. import java.sql.Timestamp
  18. /**
  19.   * 检测同一用户在一段时间内，多次登陆失败则报警行为
  20.   * 改进版
  21.   */
  22. object LoginCheckAnalysis {
  24.   def main(args: Array[String]): Unit = {
  26.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  27.     env.setParallelism(1)
  28.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  30.     val inputStream: DataStream[LoginEventData] = env.addSource(new LoginEventDataSource)
  31.         .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEventData](Time.seconds(2)) {
  32.           override def extractTimestamp(element: LoginEventData): Long = element.timestamp
  33.         })
  35.     //进行判断和检测,如果两秒内登陆失败则输出报警信息
  36.     val loginFailWarningStream: DataStream[LoginFailWarning] = inputStream.keyBy(_.userId)
  37.       .process(new LoginFailWarningAdvanceResult(2))
  39.     loginFailWarningStream.map(data=>{
  40.       (data.userId , new Timestamp(data.firstFailTime).toString , new Timestamp(data.lastFailTime).toString , data.warningMsg)
  41.     }).print("fail")
  43.     env.execute("LoginCheckJob")
  44.   }
  45. }
  47. //处理类，改进版，当数据出现连续的失败时立即报警，而不是等待2秒后
  48. class LoginFailWarningAdvanceResult(failMaxSize: Int) extends KeyedProcessFunction[String , LoginEventData , LoginFailWarning]{
  50.   //保存所有登陆失败事件
  51.   lazy val loginFailListState: ListState[LoginEventData] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEventData]("failEvent-list",classOf[LoginEventData]))
  53.   //每来一条数据，对应调用一次processElement
  54.   override def processElement(value: LoginEventData, ctx: KeyedProcessFunction[String, LoginEventData, LoginFailWarning]#Context, out: Collector[LoginFailWarning]): Unit = {
  55. //    println("数据进入process = "+value)
  57.     //判断是否失败事件
  58.     if (value.eventType == "fail"){
  60.       val iter: util.Iterator[LoginEventData] = loginFailListState.get().iterator()
  62.       if (iter.hasNext){
  63.         val preData: LoginEventData = iter.next()
  65.         //如果和之前失败的时间间隔在2秒内，则报警输出
  66.         if (value.timestamp < preData.timestamp+2000L){
  67.           out.collect(new LoginFailWarning(value.userId ,
  68.             preData.timestamp, // 第一个元素
  69.             value.timestamp, // 最后一个元素
  70.             "login fail for 2 times"
  71.           ))
  72.         }
  73.         //不管有没有在2秒内间隔的失败都将状态清空，再以最新的数据作为下一次失败的基准
  74.         loginFailListState.clear()
  75.         loginFailListState.add(value)
  76.       }else{
  77.         //之前没有失败的数据，直接添加到状态中
  78.         loginFailListState.add(value)
  79.       }
  80.     }else{
  81.       //如果是成功，直接清空状态 和 定时器
  82.       loginFailListState.clear()
  83.     }
  84.   }
  85. }

5.6.3、CEP版本

  1. <!-- cep库相关依赖 -->        
  2. <dependency>
  3.     <groupId>org.apache.flink</groupId>
  4.     <artifactId>flink-cep-scala_${scala.binary.version}</artifactId>
  5.     <version>${flink.version}</version>
  6. </dependency>
  1. package com.learn.loginfail
  3. import java.sql.Timestamp
  4. import java.util
  6. import org.apache.flink.cep.PatternSelectFunction
  7. import org.apache.flink.streaming.api.TimeCharacteristic
  8. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  9. import org.apache.flink.streaming.api.scala._
  10. import org.apache.flink.streaming.api.windowing.time.Time
  11. import org.apache.flink.cep.scala.{CEP, PatternStream}
  12. import org.apache.flink.cep.scala.pattern.Pattern
  15. /**
  16.   * 检测同一用户在一段时间内，多次登陆失败则报警行为
  17.   * 改进版
  18.   */
  19. object LoginCheckAnalysisCEP {
  21.   def main(args: Array[String]): Unit = {
  23.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  24.     env.setParallelism(1)
  25.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  27.     val inputStream: DataStream[LoginEventData] = env.addSource(new LoginEventDataSource)
  28.         .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEventData](Time.seconds(2)) {
  29.           override def extractTimestamp(element: LoginEventData): Long = element.timestamp
  30.         })
  32.     //1、定义匹配模式,连续两秒内登陆失败
  33.     val loginFailPattern: Pattern[LoginEventData, LoginEventData] = Pattern
  34.       .begin[LoginEventData]("firstFail").where(_.eventType=="fail")
  35.       .next("secondFail").where(_.eventType=="fail")
  36.       .within(Time.seconds(2))//在2秒内
  39.     //2、将模式应用到数据流上，得到一个PatternStream
  40.     val patternStream: PatternStream[LoginEventData] = CEP.pattern(inputStream.keyBy(_.userId),loginFailPattern)
  42.     //3、检出符合模式的数据流，需要调用select，(将检测出的一组数据放到一个map里面进行操作)
  43.     val result: DataStream[LoginFailWarning] = patternStream.select(new LoginFailEventMatch)
  45.     result.map(data=>{
  46.       (data.userId , new Timestamp(data.firstFailTime).toString , new Timestamp(data.lastFailTime).toString , data.warningMsg)
  47.     }).print("fail")
  49.     env.execute("LoginCheckJob")
  50.   }
  52. }
  54. /**
  55.   * 实现自定义PatternSelectFunction
  56.   */
  57. class LoginFailEventMatch extends PatternSelectFunction[LoginEventData , LoginFailWarning]{
  58.   override def select(map: util.Map[String, util.List[LoginEventData]]): LoginFailWarning = {
  59.     val firstFail: LoginEventData = map.get("firstFail").get(0)
  60.     val secondFail: LoginEventData = map.get("secondFail").get(0)
  61.     LoginFailWarning(firstFail.userId , firstFail.timestamp , secondFail.timestamp , "login fail")
  62.   }
  63. }