Flink在用户行为分析中的应用(一)

//获取状态state中的数据迭代器
      //todo 这里只能get,不能上面传入迭代器,否则迭代一次,就不能迭代第二次了
      Iterable<EventBean> eventBean = eventBeanListState.get();
      //获取事件组合条件中的感兴趣的事件
      List<EventCondition> eventConditionList = combinationCondition.getEventConditionList();
      StringBuilder sb = new StringBuilder();
      for (EventBean bean : eventBean) {
          if (bean.getTimeStamp() > queryRangeStart && bean.getTimeStamp() < queryRangeEnd && deviceId.equals(bean.getDeviceId())){
              //判断当前迭代到的bean,是否是条件中感兴趣的事件
              for (int i = 1; i < eventConditionList.size(); i++) {
                  //条件中感兴趣的事件与bean进行对比
                 if (EventUtil.eventMatchCondition(bean,eventConditionList.get(i-1)) ) {
                     sb.append(i);
                     //只要匹配到就结束
                     break;
                 }
              }

          }
      }

return sb.toString();


//获取匹配到的事件
String conditionStr = getEventCombinationConditionStr(deviceId, combinationCondition, queryRangeStart, queryRangeEnd);
//封装正则匹配的逻辑
int cnt = EventUtil.sequenceStrMatchRegexCount(conditionStr, combinationCondition.getMatchPattern());
//匹配到的步骤数
return cnt;

//遍历ck返回结果
     StringBuilder sb = new StringBuilder();
     while (resultSet.next()) {
         String eventId = resultSet.getString(1);
         //统一正则表达式
         //indexOf 从o开始算
         sb.append((eventIds.indexOf(eventId) + 1));
     }
return sb.toString();

//然后取出组合条件中的正则表达式,匹配规则

public void onTimer(long timestamp, KeyedProcessFunction<String, EventBean, RuleMatchResult>.OnTimerContext ctx, Collector<RuleMatchResult> out) throws Exception {
    Iterable<Tuple2<MarketingRule, Long>> ruleTimerIterable = ruleTimerState.get();
    Iterator<Tuple2<MarketingRule, Long>> ruleTimerIterator = ruleTimerIterable.iterator();
    while (ruleTimerIterator.hasNext()) {
        Tuple2<MarketingRule, Long> tp = ruleTimerIterator.next();
        //判断这个(规则:定时点),是否对应本次的触发点
          if (tp.f1 == timestamp){
              //如果对应,检查该规则的定时条件(定时条件中包含的就是行为条件列表)
              TimerCondition timerCondition = tp.f0.getTimerConditionList().get(0);
              //调用service去检查在条件指定的时间范围内,事件的组合发生次数是否满足
              boolean b = controller.isMatchTimeCondition(ctx.getCurrentKey(), timerCondition, timestamp - timerCondition.getTimeLate(), timestamp);
              //清理已经检查完毕的规则定时点state信息
              ruleTimerIterator.remove();
              if (b){
                out.collect(new RuleMatchResult(ctx.getCurrentKey(),tp.f0.getRuleId(),timestamp,System.currentTimeMillis()));
              }
          }
          //todo ADD 增加删除过期定时信息的逻辑
          if (tp.f1 < timestamp){
               ruleTimerIterator.remove();
          }
    }
}

//set中是去重后的业务方规定的规则
for (String keyByFields : set) {
    StringBuilder sb = new StringBuilder();
    String[] fieldNames = keyByFields.split(",");
    for (String fieldName : fieldNames) {
        //java反射获取类型对应的具体值
        Class<?> name = Class.forName("marketing.beans.EventBean");
        Field field = name.getDeclaredField(fieldName);
        //设置权限
        field.setAccessible(true);
        String fieldValue = (String)field.get(eventBean);
        sb.append(fieldValue).append(",");

    }
    String keyByValue = sb.toString().substring(0, sb.length() - 1);
    eventBean.setKeyByValue(keyByValue);
    //每种分组规则都要进行分发,相当于数据扩大了N倍
    collector.collect(new DynamicKeyedBean(keyByValue,keyByFields,eventBean));
}