Flink难点：彻底明白CEP8，CEP库中的时间、例子、版本说明

问题导读

1.对于事件延迟，cep是如何处理的?
2.对于cep编程，本文举了一个什么例子？
3.从较旧的Flink版本迁移有哪些需要注意？

上一篇：
Flink难点：彻底明白CEP8，模式检测
http://www.aboutyun.com/forum.php?mod=viewthread&tid=27335


CEP库中的时间

处理事件时间的延迟
在CEP中，元素处理的顺序很重要。 为了保证在事件时间工作时元素以正确的顺序处理，传入元素最初放在缓冲区中，元素按照时间戳按升序排序，当水印到达时，此缓冲区中的所有元素都包含在 处理小于水印的时间戳。 这意味着水印之间的元素按事件时间顺序处理。

注意：在事件时间工作时，库假定水印的正确性。

为了保证水印中的元素按事件时间顺序处理，Flink的CEP库假定水印的正确性，并将其视为时间戳小于上次看到的水印的后期元素。 后期元素不会被进一步处理。 此外，可以指定sideOutput标记来收集最后看到的水印之后的后期元素，可以像这样使用它。


[mw_shl_code=scala,true]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)[/mw_shl_code]
[mw_shl_code=java,true]PatternStream<Event> patternStream = CEP.pattern(input, pattern);

OutputTag<String> lateDataOutputTag = new OutputTag<String>("late-data"){};

SingleOutputStreamOperator<ComplexEvent> result = patternStream
    .sideOutputLateData(lateDataOutputTag)
    .select(
        new PatternSelectFunction<Event, ComplexEvent>() {...}
    );

DataStream<String> lateData = result.getSideOutput(lateDataOutputTag);[/mw_shl_code]

时间context
在PatternProcessFunction以及IterativeCondition中，用户可以访问实现TimeContext的上下文，如下所示：
[mw_shl_code=scala,true]/**
* 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();
}[/mw_shl_code]
此上下文使用户可以访问已处理事件的时间特征（IterativeCondition中的传入记录和PatternProcessFunction情况下的匹配）。 调用TimeContext＃currentProcessingTime总是为提供当前处理时间的值，此调用应优先于例如 调用System.currentTimeMillis（）。

在TimeContext #timestamp（）的情况下，返回的值等于EventTime的指定时间戳。 在ProcessingTime中，这将等于所述事件进入cep运算符的时间点（或者在PatternProcessFunction的情况下生成匹配时）。 这意味着该值在多次调用该方法时将保持一致。


例子
以下示例检测事件的key数据流上的模式start，middle（name =“error”） - > end（name =“critical”）。 事件由其ID键入，并且有效模式必须在10秒内发生。 整个处理是在事件时间完成的。
[mw_shl_code=scala,true]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(_))[/mw_shl_code]
[mw_shl_code=java,true]StreamExecutionEnvironment env = ...
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

DataStream<Event> input = ...

DataStream<Event> partitionedInput = input.keyBy(new KeySelector<Event, Integer>() {
        @Override
        public Integer getKey(Event value) throws Exception {
                return value.getId();
        }
});

Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
        .next("middle").where(new SimpleCondition<Event>() {
                @Override
                public boolean filter(Event value) throws Exception {
                        return value.getName().equals("error");
                }
        }).followedBy("end").where(new SimpleCondition<Event>() {
                @Override
                public boolean filter(Event value) throws Exception {
                        return value.getName().equals("critical");
                }
        }).within(Time.seconds(10));

PatternStream<Event> patternStream = CEP.pattern(partitionedInput, pattern);

DataStream<Alert> alerts = patternStream.select(new PatternSelectFunction<Event, Alert>() {
        @Override
        public Alert select(Map<String, List<Event>> pattern) throws Exception {
                return createAlert(pattern);
        }
});[/mw_shl_code]

从较旧的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中的followBy（）暗示了非确定性松弛连续性（见此处）。在Flink 1.3中，这已经改变并且followBy（）意味着放松的连续性，而如果需要非确定性的松弛连续性，则应该使用followAyAny（）。
  

最新经典文章，欢迎关注公众号

About云 VIP会员套餐介绍

http://www.aboutyun.com/forum.php?mod=viewthread&tid=27305

加入About云知识星球，获取更多实用资料

1.对于事件延迟，cep是如何处理的?
- 水印处理，传入元素最初放在缓冲区中，此缓冲区中的所有元素都包含在 处理小于水印的时间戳，元素按照时间戳按升序排序， 此外，可以指定sideOutput标记来收集最后看到的水印之后的后期元素。

2.对于cep编程，本文举了一个什么例子？
- 举了一个CEP系统接收一个基于EventTime的数据流，通过模式匹配保证程序有效模式在10秒内的例子

3.从较旧的Flink版本迁移有哪些需要注意？
- 迁移到1.4+：CEP库与<= Flink 1.2的向后兼容性被删除。无法恢复曾经使用1.2.x运行的CEP作业
- 迁移到1.3.x.：
    更改where条件；
    更改select（...）和flatSelect（...）方法的参数；
    Flink 1.1和1.2中的followBy（）改成followAyAny（）