flink CEP（Complex event processing），是在Flink之上實(shí)現(xiàn)的復(fù)雜事件處理庫(kù)，可以允許我們?cè)诓粩嗟牧魇綌?shù)據(jù)中通過(guò)我們自己定義的模式（Pattern）檢測(cè)和獲取出我們想要的數(shù)據(jù)，然后對(duì)這些數(shù)據(jù)進(jìn)行下一步的處理。通過(guò)各種pattern的組合，我們可以定義出非常復(fù)雜的模式來(lái)匹配我們的數(shù)據(jù)。

網(wǎng)上講CEP原理和用法的文章很多，大家可以參考下 https://juejin.im/post/5de1f32af265da05cc3190f9#heading-9

簡(jiǎn)單來(lái)說(shuō)一下，其實(shí)我們可以把使用flink cep當(dāng)做我們平時(shí)用的正則表達(dá)式，cep中的Pattern就是我們定義的正則表達(dá)式，flink中的DataStream就是正則表達(dá)式中待匹配的字符串，flink 通過(guò)DataStream 和 自定義的Pattern進(jìn)行匹配，生成一個(gè)經(jīng)過(guò)過(guò)濾之后的DataStream .

基于自定義的pattern，我們可以做很多工作，比如監(jiān)控報(bào)警、風(fēng)控、反爬等等，接下來(lái)我們基于一個(gè)簡(jiǎn)單的報(bào)警小例子來(lái)講解一些FLINK cep的實(shí)際應(yīng)用。

案例詳解

我們基于flink CEP做一個(gè)簡(jiǎn)單的報(bào)警，首先我們簡(jiǎn)化一下報(bào)警的需求
1.統(tǒng)計(jì)出來(lái)每秒鐘http狀態(tài)碼為非200的數(shù)量所占比例。大于0.7的時(shí)候觸發(fā)報(bào)警。
2.統(tǒng)計(jì)結(jié)果連續(xù)發(fā)生三大于閾值（0.7，這個(gè)數(shù)字是我自己寫(xiě)的，為了測(cè)試用，真實(shí)環(huán)境需要根據(jù)實(shí)際經(jīng)驗(yàn)來(lái)設(shè)置）發(fā)送報(bào)警通知。
3.統(tǒng)計(jì)結(jié)果小于等于閾值觸發(fā)報(bào)警恢復(fù)通知。

實(shí)際應(yīng)用中我們一般會(huì)去消費(fèi)kafka的數(shù)據(jù)來(lái)作為source、這里我們?yōu)榱撕?jiǎn)化，通過(guò)自定義source生成一些模擬的數(shù)據(jù)。

 public static class MySource implements SourceFunction<Tuple4<String,Long,Integer,Integer>>{

  static int status[] = {200, 404, 500, 501, 301};

  @Override
  public void run(SourceContext<Tuple4<String,Long,Integer,Integer>> sourceContext) throws Exception{
   while (true){
    Thread.sleep((int) (Math.random() * 100));
    // traceid,timestamp,status,response time

    Tuple4 log = Tuple4.of(
      UUID.randomUUID().toString(),
      System.currentTimeMillis(),
      status[(int) (Math.random() * 4)],
      (int) (Math.random() * 100));

    sourceContext.collect(log);
   }
  }

  @Override
  public void cancel(){

  }
 }

接下來(lái)我們定義一個(gè)sql，用來(lái)計(jì)算我們的需求中的第一個(gè)要求。

  String sql = "select pv,errorcount,round(CAST(errorcount AS DOUBLE)/pv,2) as errorRate," +
               "(starttime + interval '8' hour ) as stime," +
               "(endtime + interval '8' hour ) as etime  " +
               "from (select count(*) as pv," +
               "sum(case when status = 200 then 0 else 1 end) as errorcount, " +
               "TUMBLE_START(proctime,INTERVAL '1' SECOND)  as starttime," +
               "TUMBLE_END(proctime,INTERVAL '1' SECOND)  as endtime  " +
               "from log  group by TUMBLE(proctime,INTERVAL '1' SECOND) )";

通過(guò)執(zhí)行sql，我們獲取到了一個(gè)Result對(duì)象的DataStream,

  Table table = tenv.sqlQuery(sql);
  DataStream<Result> ds1 = tenv.toAppendStream(table, Result.class);

接下來(lái)我們到了最核心的地方，我們需要定一個(gè)Pattern。

  Pattern pattern = Pattern.<Result>begin("alert").where(new IterativeCondition<Result>(){
   @Override
   public boolean filter(
     Result i, Context<Result> context) throws Exception{
    return i.getErrorRate() > 0.7D;
   }
  }).times(3).consecutive().followedBy("recovery").where(new IterativeCondition<Result>(){
   @Override
   public boolean filter(
     Result i,
     Context<Result> context) throws Exception{
    return i.getErrorRate() <= 0.7D;
   }
  }).optional();

來(lái)詳細(xì)解釋一下這個(gè)Pattern

1. 首先定義一個(gè)名為alert的Pattern，該P(yáng)attern的作用就是過(guò)濾出錯(cuò)誤率大于0.7的數(shù)據(jù)，
2. times(3) ,表示要匹配三次，也就是要三次大于0.7.
3. consecutive 表示上述的三次匹配要是連續(xù)的，比如0.75、0.8、0.78，只有類似這樣的數(shù)據(jù)才能被匹配到，中間不能有不符合的數(shù)據(jù)出現(xiàn)。
4. followedBy表示該alert pattern的下面要跟著一個(gè)recovery pattern，而followedBy是寬松匹配，也就是兩個(gè)模式之間可以有其他的數(shù)據(jù)，如果要采用嚴(yán)格匹配，是使用next.
5. 最后recovery pattern加上一個(gè)optional 是我為了區(qū)分報(bào)警，和報(bào)警恢復(fù)想的的一個(gè)方案，這樣的話，如果是只匹配到了alert pattern，輸出的就是報(bào)警，如果recovery pattern也匹配到了，那么就是報(bào)警恢復(fù)。

在我們獲得了相應(yīng)的報(bào)警和恢復(fù)之后，接下來(lái)就是調(diào)用報(bào)警接口進(jìn)行處理了，我們這只是簡(jiǎn)單的打印出來(lái)信息。

  DataStream<Map<String,List<Result>>> alertStream = org.apache.flink.cep.CEP.pattern(
    ds1,
    pattern).select(new PatternSelectFunction<Result,Map<String,List<Result>>>(){
   @Override
   public Map<String,List<Result>> select(Map<String,List<Result>> map) throws Exception{
    List<Result> alertList = map.get("alert");
    List<Result> recoveryList = map.get("recovery");

    if (recoveryList != null){
     System.out.print("接受到了報(bào)警恢復(fù)的信息，報(bào)警信息如下：");
     System.out.print(alertList);
     System.out.print("  對(duì)應(yīng)的恢復(fù)信息：");
     System.out.println(recoveryList);
    } else {
     System.out.print("收到了報(bào)警信息 ");
     System.out.print(alertList);
    }

    return map;
   }
  });