Apache FlinkCEP 实现超时状态监控的步骤详解

 public static  SingleOutputStreamOperator  createPatternStream(...){...} public static  SingleOutputStreamOperator  createTimeoutPatternStream(...){...} final SingleOutputStreamOperator  patternStream;

 @Public public class SingleOutputStreamOperator  extends DataStream  {...}

 PatternStream ( final DataStream  inputStream, final Pattern  pattern) { this .inputStream = inputStream; this .pattern = pattern; this .comparator = null ; } PatternStream ( final DataStream  inputStream, final Pattern  pattern, final EventComparator  comparator) { this .inputStream = inputStream; this .pattern = pattern; this .comparator = comparator; }

 public class Pattern  { /** 模式名称 */ private final String name; /** 前面一个模式 */ private final Pattern  previous; /** 一个事件如果要被当前模式匹配到，必须满足的约束条件 */ private IterativeCondition  condition; /** 时间窗口长度，在时间长度内进行模式匹配 */ private Time windowTime; /** 模式量词，意思是一个模式匹配几个事件等 默认是匹配到一个 */ private Quantifier quantifier = Quantifier .one( ConsumingStrategy .STRICT); /** 停止将事件收集到循环状态时，事件必须满足的条件 */ private IterativeCondition  untilCondition; /** * 适用于{@code times}模式，用来维护模式里事件可以连续发生的次数 */ private Times times; // 匹配到事件之后的跳过策略 private final AfterMatchSkipStrategy afterMatchSkipStrategy; ... }

 public class Quantifier { ... /** * 5个属性，可以组合，但并非所有的组合都是有效的 */ public enum QuantifierProperty { SINGLE, LOOPING, TIMES, OPTIONAL, GREEDY } /** * 描述在此模式中匹配哪些事件的策略 */ public enum ConsumingStrategy { STRICT, SKIP_TILL_NEXT, SKIP_TILL_ANY, NOT_FOLLOW, NOT_NEXT } /** * 描述当前模式里事件可以连续发生的次数；举个例子，模式条件无非就是boolean，满足true条件的事件连续出现times次，或者一个次数范围，比如2~4次，2次，3次，4次都会被当前模式匹配出来，因此同一个事件会被重复匹配到 */ public static class Times { private final int from; private final int to; private Times ( int from, int to) { Preconditions .checkArgument(from > 0 , "The from should be a positive number greater than 0." ); Preconditions .checkArgument(to >= from, "The to should be a number greater than or equal to from: " + from + "." ); this .from = from; this .to = to; } public int getFrom() { return from; } public int getTo() { return to; } // 次数范围 public static Times of( int from, int to) { return new Times (from, to); } // 指定具体次数 public static Times of( int times) { return new Times (times, times); } @Override public boolean equals( Object o) { if ( this == o) { return true ; } if (o == null || getClass() != o.getClass()) { return false ; } Times times = ( Times ) o; return from == times.from && to == times.to; } @Override public int hashCode() { return Objects .hash(from, to); } } ... }

 public interface EventComparator  extends Comparator , Serializable { long serialVersionUID = 1L ; }

 public class NFACompiler { ... /** * NFAFactory 创建NFA的接口 * * @param  Type of the input events which are processed by the NFA */ public interface NFAFactory  extends Serializable { NFA createNFA(); } /** * NFAFactory的具体实现NFAFactoryImpl * * 
The implementation takes the input type serializer, the window time and the set of * states and their transitions to be able to create an NFA from them. * * @param  Type of the input events which are processed by the NFA */ private static class NFAFactoryImpl  implements NFAFactory  { private static final long serialVersionUID = 8939783698296714379L ; private final long windowTime; private final Collection > states; private final boolean timeoutHandling; private NFAFactoryImpl ( long windowTime, Collection > states, boolean timeoutHandling) { this .windowTime = windowTime; this .states = states; this .timeoutHandling = timeoutHandling; } @Override public NFA createNFA() { // 一个NFA由状态集合、时间窗口的长度和是否处理超时组成 return new NFA<>(states, windowTime, timeoutHandling); } } }

 public class NFA { /** * NFACompiler返回的所有有效的NFA状态集合 * These are directly derived from the user-specified pattern. */ private final Map > states; /** * Pattern.within(Time)指定的时间窗口长度 */ private final long windowTime; /** * 一个超时匹配的标记 */ private final boolean handleTimeout; ... }

 public interface PatternSelectFunction  extends Function , Serializable { /** * 从给到的事件映射中生成一个结果。这些事件使用他们关联的模式名称作为唯一标识 */ OUT select( Map > pattern) throws Exception ; }

 public interface PatternFlatSelectFunction  extends Function , Serializable { /** * 生成一个或多个结果 */ void flatSelect( Map > pattern, Collector  out) throws Exception ; }

 public class SelectTimeoutCepOperator  extends AbstractKeyedCEPPatternOperator > { private OutputTag  timedOutOutputTag; public SelectTimeoutCepOperator ( TypeSerializer  inputSerializer, boolean isProcessingTime, NFACompiler . NFAFactory  nfaFactory, final EventComparator  comparator, AfterMatchSkipStrategy skipStrategy, // 参数命名混淆了flat...包括SelectWrapper类中的成员命名... PatternSelectFunction  flatSelectFunction, PatternTimeoutFunction  flatTimeoutFunction, OutputTag  outputTag, OutputTag  lateDataOutputTag) { super ( inputSerializer, isProcessingTime, nfaFactory, comparator, skipStrategy, new SelectWrapper <>(flatSelectFunction, flatTimeoutFunction), lateDataOutputTag); this .timedOutOutputTag = outputTag; } ... } public interface PatternTimeoutFunction  extends Function , Serializable { OUT timeout( Map > pattern, long timeoutTimestamp) throws Exception ; } public interface PatternFlatTimeoutFunction  extends Function , Serializable { void timeout( Map > pattern, long timeoutTimestamp, Collector  out) throws Exception ; }

 public class CEP { public static  PatternStream  pattern( DataStream  input, Pattern  pattern) { return new PatternStream <>(input, pattern); } public static  PatternStream  pattern( DataStream  input, Pattern  pattern, EventComparator  comparator) { return new PatternStream <>(input, pattern, comparator); } }

 public class CEPOperatorUtils { ... private static  SingleOutputStreamOperator  createPatternStream( final DataStream  inputStream, final Pattern  pattern, final TypeInformation  outTypeInfo, final boolean timeoutHandling, final EventComparator  comparator, final OperatorBuilder  operatorBuilder) { final TypeSerializer  inputSerializer = inputStream.getType().createSerializer(inputStream.getExecutionConfig()); // check whether we use processing time final boolean isProcessingTime = inputStream.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic . ProcessingTime ; // compile our pattern into a NFAFactory to instantiate NFAs later on final NFACompiler . NFAFactory  nfaFactory = NFACompiler .compileFactory(pattern, timeoutHandling); final SingleOutputStreamOperator  patternStream; if (inputStream instanceof KeyedStream ) { KeyedStream  keyedStream = ( KeyedStream ) inputStream; patternStream = keyedStream.transform( operatorBuilder.getKeyedOperatorName(), outTypeInfo, operatorBuilder.build( inputSerializer, isProcessingTime, nfaFactory, comparator, pattern.getAfterMatchSkipStrategy())); } else { KeySelector  keySelector = new NullByteKeySelector <>(); patternStream = inputStream.keyBy(keySelector).transform( operatorBuilder.getOperatorName(), outTypeInfo, operatorBuilder.build( inputSerializer, isProcessingTime, nfaFactory, comparator, pattern.getAfterMatchSkipStrategy() )).forceNonParallel(); } return patternStream; } ... }

 KeySelector  keySelector = new NullByteKeySelector <>();

 public class CEPTimeoutEventJob { private static final String LOCAL_KAFKA_BROKER = "localhost:9092" ; private static final String GROUP_ID = CEPTimeoutEventJob . class .getSimpleName(); private static final String GROUP_TOPIC = GROUP_ID; public static void main( String [] args) throws Exception { // 参数 ParameterTool params = ParameterTool .fromArgs(args); StreamExecutionEnvironment env = StreamExecutionEnvironment .getExecutionEnvironment(); // 使用事件时间 env.setStreamTimeCharacteristic( TimeCharacteristic . EventTime ); env.enableCheckpointing( 5000 ); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig . ExternalizedCheckpointCleanup .RETAIN_ON_CANCELLATION); env.getConfig().disableSysoutLogging(); env.getConfig().setRestartStrategy( RestartStrategies .fixedDelayRestart( 5 , 10000 )); // 不使用POJO的时间 final AssignerWithPeriodicWatermarks extractor = new IngestionTimeExtractor (); // 与Kafka Topic的Partition保持一致 env.setParallelism( 3 ); Properties kafkaProps = new Properties (); kafkaProps.setProperty( "bootstrap.servers" , LOCAL_KAFKA_BROKER); kafkaProps.setProperty( "group.id" , GROUP_ID); // 接入Kafka的消息 FlinkKafkaConsumer011  consumer = new FlinkKafkaConsumer011 <>(GROUP_TOPIC, new POJOSchema (), kafkaProps); DataStream  pojoDataStream = env.addSource(consumer) .assignTimestampsAndWatermarks(extractor); pojoDataStream.print(); // 根据主键aid分组 即对每一个POJO事件进行匹配检测【不同类型的POJO，可以采用不同的within时间】 // 1. DataStream  keyedPojos = pojoDataStream .keyBy( "aid" ); // 从初始化到终态-一个完整的POJO事件序列 // 2. Pattern  completedPojo = Pattern .begin( "init" ) .where( new SimpleCondition () { private static final long serialVersionUID = - 6847788055093903603L ; @Override public boolean filter(POJO pojo) throws Exception { return "02" .equals(pojo.getAstatus()); } }) .followedBy( "end" ) //            .next("end") .where( new SimpleCondition () { private static final long serialVersionUID = - 2655089736460847552L ; @Override public boolean filter(POJO pojo) throws Exception { return "00" .equals(pojo.getAstatus()) || "01" .equals(pojo.getAstatus()); } }); // 找出1分钟内【便于测试】都没有到终态的事件aid // 如果针对不同类型有不同within时间，比如有的是超时1分钟，有的可能是超时1个小时 则生成多个PatternStream // 3. PatternStream  patternStream = CEP.pattern(keyedPojos, completedPojo.within( Time .minutes( 1 ))); // 定义侧面输出timedout // 4. OutputTag  timedout = new OutputTag ( "timedout" ) { private static final long serialVersionUID = 773503794597666247L ; }; // OutputTag timeoutOutputTag, PatternFlatTimeoutFunction patternFlatTimeoutFunction, PatternFlatSelectFunction patternFlatSelectFunction // 5. SingleOutputStreamOperator  timeoutPojos = patternStream.flatSelect( timedout, new POJOTimedOut (), new FlatSelectNothing () ); // 打印输出超时的POJO // 6.7. timeoutPojos.getSideOutput(timedout).print(); timeoutPojos.print(); env.execute( CEPTimeoutEventJob . class .getSimpleName()); } /** * 把超时的事件收集起来 */ public static class POJOTimedOut implements PatternFlatTimeoutFunction  { private static final long serialVersionUID = - 4214641891396057732L ; @Override public void timeout( Map > map, long l, Collector  collector) throws Exception { if ( null != map.get( "init" )) { for (POJO pojoInit : map.get( "init" )) { System .out.println( "timeout init:" + pojoInit.getAid()); collector.collect(pojoInit); } } // 因为end超时了，还没收到end，所以这里是拿不到end的 System .out.println( "timeout end: " + map.get( "end" )); } } /** * 通常什么都不做，但也可以把所有匹配到的事件发往下游；如果是宽松临近，被忽略或穿透的事件就没办法选中发往下游了 * 一分钟时间内走完init和end的数据 * * @param  */ public static class FlatSelectNothing  implements PatternFlatSelectFunction  { private static final long serialVersionUID = - 3029589950677623844L ; @Override public void flatSelect( Map > pattern, Collector  collector) { System .out.println( "flatSelect: " + pattern); } } }

 3 > POJO{aid= 'ID000-0' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419728242 , energy= 529.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-1' , astyle= 'STYLE000-2' , aname= 'NAME-1' , logTime= 1563419728783 , energy= 348.00 , age= 26 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-0' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419749259 , energy= 492.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '00' , createTime= null , updateTime= null } flatSelect: {init=[POJO{aid= 'ID000-0' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419728242 , energy= 529.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null }], end =[POJO{aid= 'ID000-0' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419749259 , energy= 492.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '00' , createTime= null , updateTime= null }]} timeout init:ID000- 1 3 > POJO{aid= 'ID000-1' , astyle= 'STYLE000-2' , aname= 'NAME-1' , logTime= 1563419728783 , energy= 348.00 , age= 26 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null } timeout end : null 3 > POJO{aid= 'ID000-2' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419829639 , energy= 467.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '03' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-2' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419841394 , energy= 107.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '00' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-3' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419967721 , energy= 431.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-3' , astyle= 'STYLE000-2' , aname= 'NAME-0' , logTime= 1563419979567 , energy= 32.00 , age= 26 , tt= 2019 - 07 - 18 , astatus= '03' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-3' , astyle= 'STYLE000-2' , aname= 'NAME-0' , logTime= 1563419993612 , energy= 542.00 , age= 26 , tt= 2019 - 07 - 18 , astatus= '01' , createTime= null , updateTime= null } flatSelect: {init=[POJO{aid= 'ID000-3' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563419967721 , energy= 431.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null }], end =[POJO{aid= 'ID000-3' , astyle= 'STYLE000-2' , aname= 'NAME-0' , logTime= 1563419993612 , energy= 542.00 , age= 26 , tt= 2019 - 07 - 18 , astatus= '01' , createTime= null , updateTime= null }]} 3 > POJO{aid= 'ID000-4' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563420063760 , energy= 122.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null } 3 > POJO{aid= 'ID000-4' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563420078008 , energy= 275.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '03' , createTime= null , updateTime= null } timeout init:ID000- 4 3 > POJO{aid= 'ID000-4' , astyle= 'STYLE000-0' , aname= 'NAME-0' , logTime= 1563420063760 , energy= 122.00 , age= 0 , tt= 2019 - 07 - 18 , astatus= '02' , createTime= null , updateTime= null } timeout end : null