实时预警需求分析

淩亂°似流年 2022-11-02 13:21 250阅读 0赞

# 实时预警需求分析 #

**实时预警**，是一种经常出现在实时计算中的业务类型。根据日志数据中系统报错异常，或者用户行为异常的检测，产生对应预警日志。预警日志通过图形化界面的展示，可以提醒监控方，需要及时核查问题，并采取应对措施。

## 一、需求说明 ##

需求：同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵，并且过程中没有浏览商品。达到以上要求则产生一条预警日志。并且同一设备，每分钟只记录一次预警。

预警日志格式  
![在这里插入图片描述][20210225160011191.png]

## 二、整体流程设计 ##

框架流程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

**开发思路**

1.  从kafka中消费数据，根据条件进行过滤筛选，生成预警日志；
2.  预警日志保存到ElasticSearch中；
3.  利用Kibana快速搭建可视化图形界面。

## 三、实时计算模块 ##

**筛选条件分析**：

*  同一设备（分组）
 *  5分钟内（窗口）
 *  三次不同账号登录（用户去重）
 *  领取优惠券（行为）
 *  没有浏览商品（行为）
 *  同一设备每分钟只记录一次预警（ES去重）

数据处理流程图  
![在这里插入图片描述][20210225155340423.png]

## 四、代码开发 ##

object AlterApp { 
      /** * 分析预警需求： * * 同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵，并且过程中没有浏览商品。 * 达到以上要求则产生一条预警日志。并且同一设备，每分钟只记录一次预警 * * 同一设备，在5分钟内，用三个及以上不同账号(uid)登录并领取优惠劵，且，在登录后没有浏览商品， * 如果符合此要求，则产生一条预警日志。 * * 限制： 同一设备，每分钟只记录一次预警日志！ * * * 输入数据： 日志数据中的事件日志。 * 核心字段： evid=coupon * gmall-mocker 模拟 * * 计算过程： * ①读取输入的数据 * gmall-mocker ---> 模拟数据 ----> gmall-logger ----> kafka的GMALL_EVENTLOG * * ②封装为样例类 * 事件日志有几个字段，样例类就怎么设计！ * 多来两个字段： log_hour,log_date用于分时统计！ * * ③kafka中的一条数据是一个设备，操作的一个事件的记录 * a) 设置窗口为 5min * b) 分组 ，按照 mid 将这个批次拉取的所有的数据，按照mid进行分组 * c) 记录当前组中，这个设备一共登录了多少个uid * Set集合,保存不同的uid * * 记录uid的前提是，领取了优惠券！判断当前事件是否是一个领取优惠券的事件，如果是，记录uid! * * d) 如果这个uid 浏览了商品 evid=clickItem，此时这个mid，不预警 * e) 产生一条预警日志 * mid: * uids: 领取优惠券的uid * itemIds: 优惠券设计的商品id * events: 这个设备，所有登录的uid，发生过的事件类型 * 除了clickItem以外的四种类型 * * ts: 预警日志生成的时间 * * f) 同一设备，每分钟只记录一次预警日志！ * 同一个设备，如果在一分钟内生成了多条预警日志，es中只保留一条！ * * PUT时，只需要保证，一分钟内一个设备，多次PUT记录的id是一致的！ * 00:01:01 PUT /a/b/mid_156_2021-02-20 00:01 * 00:01:02 PUT /a/b/mid_156_2021-02-20 00:01 * * 00:02:00 PUT /a/b/mid_156_2021-02-20 00:02 * * id如何设计： mid_分钟 * 输出数据： 预警日志 */
      def main(args: Array[String]): Unit = { 
      
        val streamingContext: StreamingContext = new StreamingContext("local[*]","AlterApp",Seconds(10))
      
        //1.从kafka获取DS
        val ds: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT,streamingContext)
      
        //2.将DS[ConsumerRecord]转换为DS[StartUpLog]
        val ds1= ds.map(record => { 
          val log: EventLog = JSON.parseObject(record.value(), classOf[EventLog])
          log
        })
      
        //统计5分钟内，开窗，然后按照mid分组
        val ds2: DStream[(String, Iterable[EventLog])] = ds1.window(Minutes(5))
          .map(log => (log.mid, log))
          .groupByKey()
        
        //对5分钟内数据进行操作
        val ds3: DStream[CouponAlertInfo] = ds2.map { 
          //map为一进一出，不符合要求的数据也要返回一个对象
          case (mid, logs) => { 
            //判断这个设备是否符合预警条件，如果符合，生成预警日志
            val uids: util.HashSet[String] = new util.HashSet[String]()
            val itemIds: util.HashSet[String] = new util.HashSet[String]()
            val events: util.ArrayList[String] = new util.ArrayList[String]()
          
            //提供一个是否需要预警的标记，如果点击了商品，不需要预警
            var if_clickProduct = false
          
            //记录这个设备产生的事件
            breakable { 
              logs.foreach(log => { 
                events.add(log.evid)
                //根据日志类型，是否点击商品，判断是否领取了优惠券
                if (log.evid.equals("clickItem")) { 
                  if_clickProduct = true
                  //终止循环
                  break
                }
              
                if (log.evid.equals("coupon")) { 
                  //将领取优惠券的uid记录
                  uids.add(log.uid)
                  //记录优惠券关联的商品
                  itemIds.add(log.itemid)
                }
              
              })
            }
          
            //遍历完该设备的所有事件，判断该设备是否需要预警
            if (if_clickProduct || uids.size() < 3) { 
              null
            } else { 
              CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis())
            }
          }
        }
       
        
        //将null值过滤掉
        val ds4: DStream[CouponAlertInfo] = ds3.filter(_!=null)
        
        //打印
        ds4.print(100)
        
        //将结果写入es
        ds4.foreachRDD(rdd=>{ 
          //以分区为单位写入，节省建立连接的开销
          rdd.foreachPartition(iter=>{ 
            //每天的预警信息单独存放在一个index
            val indexName = GmallConstants.ES_INDEX_ALERT + LocalDate.now()
            
            val format = new SimpleDateFormat("yyyy-MM-dd HH:mm")
            
            val docList: List[(String, CouponAlertInfo)] = iter.toList.map(info=>(info.mid + "_" + format.format(new Date(info.ts)),info))
            
            MyESUtil.insertBulk(indexName, docList)
            
          })
        })
      
      
        streamingContext.start()
        
        streamingContext.awaitTermination()
      
      }
      
    
    }

[20210225160011191.png]: /images/20221024/7a19738b76e64dfb90e53f8909570389.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221024/9940937b1711460db26f2817c8d3d559.png
[20210225155340423.png]: /images/20221024/674cdaf3df5040d084b38cd3c84e8bef.png