与oneOrMore（）和times（）结合使用，并在匹配事件之间施加严格的连续性，

即任何不匹配的元素都会破坏匹配（如next（）所示）.

如果未应用，则使用宽松的连续性（如followBy（）所示）.

E.g. a pattern like:

Pattern.begin("start").where(_.getName().equals("c"))
  .followedBy("middle").where(_.getName().equals("a"))
                       .oneOrMore().consecutive()
  .followedBy("end1").where(_.getName().equals("b"))

将为输入序列生成以下匹配项：C D A1 A2 A3 D A4 B

连续应用: {C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}

没有连续应用: {C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}, {C A1 A2 A3 A4 B}

与oneOrMore（）和times（）结合使用，并在匹配事件之间施加不确定的宽松连续性（如followByAny（）中所示）.

如果未应用，则使用宽松的连续性（如followBy（）所示）.

E.g. a pattern like:

Pattern.begin("start").where(_.getName().equals("c"))
  .followedBy("middle").where(_.getName().equals("a"))
                       .oneOrMore().allowCombinations()
  .followedBy("end1").where(_.getName().equals("b"))

将为输入序列生成以下匹配项: C D A1 A2 A3 D A4 B

启用组合: {C A1 B}, {C A1 A2 B}, {C A1 A3 B}, {C A1 A4 B}, {C A1 A2 A3 B},

{C A1 A2 A4 B}, {C A1 A3 A4 B}, {C A1 A2 A3 A4 B}

没有启用组合: {C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}, {C A1 A2 A3 A4 B}

 也可以将模式序列定义为begin，followedBy，followedByAny和next的条件。 模式序列在逻辑上将被视为匹配条件，并且将返回GroupPattern，并且可以应用oneOrMore（），times（#ofTimes），times（＃fromTimes，＃toTimes），optional（），continuous（）， allowCombinations（）到GroupPattern。

val start: Pattern[Event, _] = Pattern.begin(
    Pattern.begin[Event]("start").where(...).followedBy("start_middle").where(...)
// strict contiguity
val strict: Pattern[Event, _] = start.next(
    Pattern.begin[Event]("next_start").where(...).followedBy("next_middle").where(...)
).times(3)
// relaxed contiguity
val relaxed: Pattern[Event, _] = start.followedBy(
    Pattern.begin[Event]("followedby_start").where(...).followedBy("followedby_middle").where(...)
).oneOrMore()
// non-deterministic relaxed contiguity
val nonDetermin: Pattern[Event, _] = start.followedByAny(
    Pattern.begin[Event]("followedbyany_start").where(...).followedBy("followedbyany_middle").where(...)
).optional()

定义开始模式:

val start = Pattern.begin(
    Pattern.begin[Event]("start").where(...).followedBy("middle").where(...)
next(#name)
追加一个新模式。 匹配事件必须直接接在先前的匹配事件之后（严格连续）:
val next = start.next("middle")
next(#pattern_sequence)
追加一个新模式。 一系列匹配事件必须直接接续先前的匹配事件（严格连续）:
val next = start.next(
    Pattern.begin[Event]("start").where(...).followedBy("middle").where(...)
followedBy(#name)
追加一个新模式。 匹配事件和上一个匹配事件之间可能会发生其他事件（宽松的连续性）:
val followedBy = start.followedBy("middle")
followedBy(#pattern_sequence)
追加一个新模式。 在一系列匹配事件和上一个匹配事件之间可能会发生其他事件（宽松的连续性）:
val followedBy = start.followedBy(
    Pattern.begin[Event]("start").where(...).followedBy("middle").where(...)
followedByAny(#name)
追加一个新模式。 在匹配事件和先前的匹配事件之间可能会发生其他事件，
并且将为每个替代匹配事件显示替代匹配（不确定性宽松邻接）:
val followedByAny = start.followedByAny("middle")
followedByAny(#pattern_sequence)
追加一个新模式。 在匹配事件序列和先前的匹配事件之间可能会发生其他事件，
并且将为匹配事件的每个替代序列（非确定性宽松连续性）提供替代匹配：
val followedByAny = start.followedByAny(
    Pattern.begin[Event]("start").where(...).followedBy("middle").where(...)
notNext()
附加新的否定模式。 匹配（否定）事件必须直接继承先前的匹配事件（严格连续性）才能放弃部分匹配:
val notNext = start.notNext("not")
notFollowedBy()
附加新的否定模式。 即使在匹配（负）事件和上一个匹配事件（宽松的邻接）之间发生其他事件，部分匹配事件序列也将被丢弃:
val notFollowedBy = start.notFollowedBy("not")
within(time)
定义事件序列与模式匹配的最大时间间隔。 如果未完成的事件序列超过此时间，则将其丢弃:
pattern.within(Time.seconds(10))
匹配后跳过策略 
对于给定的模式，可以将同一事件分配给多个成功的匹配。 要控制将事件分配给多少个匹配项，您需要指定一个名为AfterMatchSkipStrategy的跳过策略。 跳过策略有五种类型，列出如下：
NO_SKIP：将发出所有可能的匹配项。
SKIP_TO_NEXT：丢弃以同一事件开始的所有部分匹配，发出的匹配从事件开始。
SKIP_PAST_LAST_EVENT：丢弃匹配开始后但结束之前，开始的所有匹配。
SKIP_TO_FIRST：丢弃在匹配开始之后但在PatternName的第一个事件发生之前开始的所有匹配。
SKIP_TO_LAST：丢弃在匹配开始之后但在PatternName的最后一个事件发生之前开始的所有匹配。
请注意，在使用SKIP_TO_FIRST和SKIP_TO_LAST跳过策略时，还应指定有效的PatternName。
例如，对于给定的模式b + c和数据流b1 b2 b3 c，这四种跳过策略之间的差异如下：
Skip StrategyResultDescription
NO_SKIP
b1 b2 b3 c
b2 b3 c
b3 c
找到匹配的b1 b2 b3 c后，匹配过程将不会丢弃任何结果。
Pattern.begin("patternName", skipStrategy)

注意：对于SKIP_TO_FIRST / LAST，有两个选项可以处理在没有元素映射到指定变量时的情况。 默认情况下，将使用NO_SKIP策略。 另一个选择是在这种情况下引发异常。 可以通过以下方式启用此选项：
AfterMatchSkipStrategy.skipToFirst(patternName).throwExceptionOnMiss()
 在指定了所需的模式序列之后，是时候将其应用于输入流以检测潜在的匹配了。 要针对模式序列运行事件流，必须创建一个PatternStream。 给定一个输入流输入，一个模式模式和一个可选的比较器比较器（用于在EventTime的情况下对具有相同时间戳的事件或在同一时刻到达的事件进行排序），您可以通过调用以下代码来创建PatternStream：
val input : DataStream[Event] = ...
val pattern : Pattern[Event, _] = ...
var comparator : EventComparator[Event] = ... // optional
val patternStream: PatternStream[Event] = CEP.pattern(input, pattern, comparator)
输入流可以是键的，也可以是非键的，具体取决于您的用例。
注意：将模式应用于非键控流将导致并行度等于1的作业。
从模式中选择
 获得PatternStream之后，可以将转换应用于检测到的事件序列。 建议的实现方式是通过PatternProcessFunction。
PatternProcessFunction具有一个processMatch方法，每个匹配事件序列都会调用该方法。 它以Map <String，List <IN >>的形式接收匹配，其中键是模式序列中每个模式的名称，值是该模式所有可接受事件的列表（IN是您的类型 输入元素）。 给定模式的事件按时间戳排序。 返回每个模式的接受事件列表的原因是，当使用循环模式（例如oneToMany（）和times（））时，给定模式可能会接受多个事件。
class MyPatternProcessFunction<IN, OUT> extends PatternProcessFunction<IN, OUT> {
    @Override
    public void processMatch(Map<String, List<IN>> match, Context ctx, Collector<OUT> out) throws Exception;
        IN startEvent = match.get("start").get(0);
        IN endEvent = match.get("end").get(0);
        out.collect(OUT(startEvent, endEvent));
PatternProcessFunction允许访问Context对象。 有了它，就可以访问与时间相关的特征，例如currentProcessingTime或当前匹配的时间戳（这是分配给匹配的最后一个元素的时间戳）。 有关更多信息，请参见时间上下文。 通过这种情况，还可以将结果发送到副输出。
处理超时的部分模式
 只要某个模式具有通过inner关键字附加的窗口长度，则可能会丢弃部分事件序列，因为它们超过了窗口长度。 要对超时的部分匹配采取行动，可以使用TimedOutPartialMatchHandler接口。 该接口应该以混合样式使用。 这意味着您还可以使用PatternProcessFunction实现此接口。 TimedOutPartialMatchHandler提供了额外的processTimedOutMatch方法，将为每个超时的部分匹配调用该方法。
class MyPatternProcessFunction<IN, OUT> extends PatternProcessFunction<IN, OUT> implements TimedOutPartialMatchHandler<IN> {
    @Override
    public void processMatch(Map<String, List<IN>> match, Context ctx, Collector<OUT> out) throws Exception;
    @Override
    public void processTimedOutMatch(Map<String, List<IN>> match, Context ctx) throws Exception;
        IN startEvent = match.get("start").get(0);
        ctx.output(outputTag, T(startEvent));
注意：processTimedOutMatch不能访问主输出。 但是，仍然可以通过Context对象通过侧面输出发出结果。
Convenience API
 前面提到的PatternProcessFunction是在Flink 1.8中引入的，从那时起，它是与匹配项进行交互的推荐方法。 仍然可以使用老式的API，例如select / flatSelect，该API在内部将转换为PatternProcessFunction。
val patternStream: PatternStream[Event] = CEP.pattern(input, pattern)
val outputTag = OutputTag[String]("side-output")
val result: SingleOutputStreamOperator[ComplexEvent] = patternStream.flatSelect(outputTag){
    (pattern: Map[String, Iterable[Event]], timestamp: Long, out: Collector[TimeoutEvent]) =>
        out.collect(TimeoutEvent())
    (pattern: mutable.Map[String, Iterable[Event]], out: Collector[ComplexEvent]) =>
        out.collect(ComplexEvent())
val timeoutResult: DataStream[TimeoutEvent] = result.getSideOutput(outputTag)
CEP库中的时间 
处理事件时间的延迟
 在CEP中，处理元素的顺序很重要。 为了确保在事件时间内工作时按正确的顺序处理元素，将传入的元素最初放在缓冲区中，在该缓冲区中，元素根据其时间戳按升序排序，并且当水印到达时，该缓冲区中的所有元素 小于水印的时间戳被处理。 这意味着水印之间的元素按事件时间顺序进行处理。
注意：在事件时间内工作时，cep 库假定水印正确无误。
为了确保跨水印的元素按事件时间顺序进行处理，Flink的CEP库假定水印是正确的，并将其时间戳小于最后看到的水印的时间戳视为晚元素。 后期元素不会进一步处理。 另外，您可以指定sideOutput标签来收集在最后看到的水印之后出现的后期元素，您可以像这样使用它。
val patternStream: PatternStream[Event] = CEP.pattern(input, pattern)
val lateDataOutputTag = OutputTag[String]("late-data")
val result: SingleOutputStreamOperator[ComplexEvent] = patternStream
      .sideOutputLateData(lateDataOutputTag)
      .select{
          pattern: Map[String, Iterable[ComplexEvent]] => ComplexEvent()
val lateData: DataStream[String] = result.getSideOutput(lateDataOutputTag)
Time context
 在PatternProcessFunction和IterativeCondition中，用户可以访问实现TimeContext的上下文，如下所示：
 * Enables access to time related characteristics such as current processing time or timestamp of
 * currently processed element. Used in {@link PatternProcessFunction} and
 * {@link org.apache.flink.cep.pattern.conditions.IterativeCondition}
@PublicEvolving
public interface TimeContext {
     * Timestamp of the element currently being processed.
     * <p>In case of {@link org.apache.flink.streaming.api.TimeCharacteristic#ProcessingTime} this
     * will be set to the time when event entered the cep operator.
    long timestamp();
    /** Returns the current processing time. */
    long currentProcessingTime();
此上下文使用户可以访问已处理事件的时间特征（在IterativeCondition情况下为传入记录，在PatternProcessFunction情况下为匹配）。 调用TimeContext＃currentProcessingTime始终会为您提供当前处理时间的值，并且此调用应优先于例如 调用System.currentTimeMillis（）。
如果使用TimeContext＃timestamp（），则返回值等于使用EventTime时分配的时间戳。 在ProcessingTime中，该时间等于所述事件进入cep运算符的时间点（或在PatternProcessFunction的情况下生成匹配项的时间点）。 这意味着该值在对该方法的多次调用中将保持一致。
Examples
以下示例在事件的键控数据流上检测模式的开始，中间（名称=“错误”）->结束（名称=“严重”）。 这些事件由其ID进行键控，并且有效模式必须在10秒内发生。 整个处理过程随事件时间而定。
val env : StreamExecutionEnvironment = ...
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val input : DataStream[Event] = ...
val partitionedInput = input.keyBy(event => event.getId)
val pattern = Pattern.begin[Event]("start")
  .next("middle").where(_.getName == "error")
  .followedBy("end").where(_.getName == "critical")
  .within(Time.seconds(10))
val patternStream = CEP.pattern(partitionedInput, pattern)
val alerts = patternStream.select(createAlert(_))
从旧版本的Flink迁移（1.3版之前） 
迁移到1.4+
在Flink-1.4中，删除了CEP库与<= Flink 1.2的向后兼容性。 不幸的是，不可能恢复曾经使用1.2.x运行的CEP作业。
迁移到1.3.x
 Flink-1.3中的CEP库附带了许多新功能，这些新功能导致API发生了一些更改。 在这里，我们描述了您需要对旧的CEP作业进行的更改，以便能够使用Flink-1.3运行它们。 进行这些更改并重新编译作业后，您将能够从作业的旧版本获取的保存点恢复执行，即无需重新处理过去的数据。
所需的更改是：
更改条件（where(...)子句中的条件）以扩展SimpleCondition类，而不是实现FilterFunction接口。
更改作为select（...）和flatSelect（...）方法的参数提供的函数，以获取与每个模式关联的事件列表（Java中为List，Scala中为Iterable）。 这是因为通过添加循环模式，多个输入事件可以匹配单个（循环）模式。
Flink 1.1和1.2中的followedBy（）隐含了不确定的宽松连续性（请参见此处）。 在Flink 1.3中，此更改已更改，并且followBy（）表示宽松邻接，而如果需要非确定性宽松邻接，则应使用followByAny（）。
欢迎关注Flink菜鸟公众号，会不定期更新Flink（开发技术）相关的推文