• Flink Cep 源码分析

    复合事件处理（Complex Event Processing，CEP）是一种基于动态环境中事件流的分析技术，事件在这里通常是有意义的状态变化，通过分析事件间的关系，利用过滤、关联、聚合等技术，根据事件间的时序关系和聚合关系制定检测规则，持续地从事件流中查询出符合要求的事件序列，最终分析得到更复杂的复合事件。

             我们从一个案例开始源码分析之路。

    1．案例代码

    1. Import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    2. import org.apache.flink.api.java.tuple.Tuple3;
    3. import org.apache.flink.cep.CEP;
    4. import org.apache.flink.cep.PatternSelectFunction;
    5. import org.apache.flink.cep.PatternStream;
    6. import org.apache.flink.cep.pattern.Pattern;
    7. import org.apache.flink.streaming.api.TimeCharacteristic;
    8. import org.apache.flink.streaming.api.datastream.KeyedStream;
    9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    10.  
    11.   
    12. import java.time.Duration;
    13. import java.util.Map;
    14.  
    15.  
    16. public class FlinkCepTest {
    17.  
    18.  
    19.     public static void main(String[] args) throws Exception {
    20.  
    21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    22.  
    23.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    24.         env.setParallelism(1);
    25.  
    26.  
    27.   //        数据源
    28.  
    29.         KeyedStream<Tuple3<String, Long, String>, String> source = env.fromElements(
    30.                 new Tuple3<String, Long, String>("1001", 1656914303000L, "success")
    31.                 , new Tuple3<String, Long, String>("1001", 1656914304000L, "fail")
    32.                 , new Tuple3<String, Long, String>("1001", 1656914305000L, "fail")
    33.                 , new Tuple3<String, Long, String>("1001", 1656914306000L, "success")
    34.                 , new Tuple3<String, Long, String>("1001", 1656914307000L, "fail")
    35.                 , new Tuple3<String, Long, String>("1001", 1656914308000L, "success")
    36.                 , new Tuple3<String, Long, String>("1001", 1656914309000L, "fail")
    37.                 , new Tuple3<String, Long, String>("1001", 1656914310000L, "success")
    38.                 , new Tuple3<String, Long, String>("1001", 1656914311000L, "fail")
    39.                 , new Tuple3<String, Long, String>("1001", 1656914312000L, "fail")
    40.                 , new Tuple3<String, Long, String>("1001", 1656914313000L, "success")
    41.                 , new Tuple3<String, Long, String>("1001", 1656914314000L, "end")
    42.         ).assignTimestampsAndWatermarks(WatermarkStrategy
    43.  
    44.                 .<Tuple3<String, Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
    45.                 .withTimestampAssigner((event, timestamp) ->{
    46.                     return event.f1;
    47.                 }))
    48.  
    49.                 .keyBy(e -> e.f0);
    50.  
    51.   
    52.  
    53.         Pattern<Tuple3<String, Long, String>,?> pattern = Pattern
    54.  
    55.                 .<Tuple3<String, Long, String>>begin("begin")
    56.  
    57.                 .where(new Begincondition())
    58.  
    59.                 .followedByAny("middle")
    60.  
    61.                 .where(new Middlecondition())
    62.  
    63.                 .followedBy("end")
    64.  
    65.                 .where(new Endcondition())
    66.  
    67.                 ;
    68.  
    69.   
    70.  
    71.         //TODO 内部构建 PatternStreamBuilder 并返回 PatternStream
    72.  
    73.         PatternStream patternStream = CEP.pattern(source, pattern);
    74.  
    75.         patternStream.select(new PatternSelectFunction<Tuple3<String, Long, String>,Map>() {
    76.             @Override
    77.             public Map select(Map map) throws Exception {
    78.                 return map;
    79.             }
    80.         }).print();
    81.         env.execute("cep");
    82.     }
    83. }

    2．源码分析

    根据上述提供的案例接下来我们分几个模块进行源码解析：Pattern构建，内部包含 NFAFactory，Cepopertor等构建逻辑，数据处理，超时处理，获取数据。

    1. //TODO 内部构建 PatternStreamBuilder 并返回 PatternStream
    2. PatternStream patternStream = CEP.pattern(source, pattern);
    3.  
    4.  
    5. PatternStreamBuilder.forStreamAndPattern(inputStream, pattern);
    6.  
    7. static <IN> PatternStreamBuilder<IN> forStreamAndPattern(final DataStream<IN> inputStream, final Pattern<IN, ?> pattern) {
    8.  
    9.    return new PatternStreamBuilder<>(inputStream, pattern, TimeBehaviour.EventTime, null, null);
    10.  
    11. }

    PatternStream.select() 内部会调用 PatternStreamBuilder.build()

    1. public <R> SingleOutputStreamOperator<R> process(
    2.  
    3.       final PatternProcessFunction<T, R> patternProcessFunction,
    4.  
    5.       final TypeInformation<R> outTypeInfo) {
    6.  
    7.  
    8.  
    9.    return builder.build(
    10.  
    11.       outTypeInfo,
    12.  
    13.       builder.clean(patternProcessFunction));
    14.  
    15. }

    org.apache.flink.cep.PatternStreamBuilder#build

    1. <OUT, K> SingleOutputStreamOperator<OUT> build(
    2.  
    3.         final TypeInformation<OUT> outTypeInfo,
    4.  
    5.         final PatternProcessFunction<IN, OUT> processFunction) {
    6.  
    7.  
    8.  
    9.     checkNotNull(outTypeInfo);
    10.  
    11.     checkNotNull(processFunction);
    12.  
    13.  
    14.  
    15.     //TODO 构造序列化器
    16.  
    17.     final TypeSerializer<IN> inputSerializer =
    18.             inputStream.getType().createSerializer(inputStream.getExecutionConfig());
    19.     final boolean isProcessingTime = timeBehaviour == TimeBehaviour.ProcessingTime;
    20.  
    21.  
    22.     //TODO 判断是否是获取超时结果的 select/flatSelect
    23.  
    24.     final boolean timeoutHandling = processFunction instanceof TimedOutPartialMatchHandler;
    25.  
    26.     //TODO 构建 NFAFactory 工厂类
    27.  
    28.     // 工厂对象还包含了用户所有的State集合
    29.  
    30.     final NFACompiler.NFAFactory<IN> nfaFactory =
    31.  
    32.             NFACompiler.compileFactory(pattern, timeoutHandling);
    33.  
    34.  
    35.  
    36.     //TODO 创建 CepOperator
    37.  
    38.     final CepOperator<IN, K, OUT> operator =
    39.  
    40.             new CepOperator<>(
    41.  
    42.                     inputSerializer,
    43.  
    44.                     isProcessingTime,
    45.  
    46.                     nfaFactory,
    47.  
    48.                     comparator,
    49.  
    50.                     pattern.getAfterMatchSkipStrategy(),
    51.  
    52.                     processFunction,
    53.  
    54.                     lateDataOutputTag);
    55.  
    56.  
    57.  
    58.     final SingleOutputStreamOperator<OUT> patternStream;
    59.  
    60.     if (inputStream instanceof KeyedStream) {
    61.  
    62.         KeyedStream<IN, K> keyedStream = (KeyedStream<IN, K>) inputStream;
    63.  
    64.  
    65.  
    66.         patternStream = keyedStream.transform("CepOperator", outTypeInfo, operator);
    67.  
    68.     } else {
    69.  
    70.         KeySelector<IN, Byte> keySelector = new NullByteKeySelector<>();
    71.  
    72.  
    73.  
    74.         patternStream =
    75.  
    76.                 inputStream
    77.  
    78.                         .keyBy(keySelector)
    79.  
    80.                         .transform("GlobalCepOperator", outTypeInfo, operator)
    81.  
    82.                         .forceNonParallel();
    83.  
    84.     }
    85.  
    86.  
    87.  
    88.     return patternStream;
    89.  
    90. }

    org.apache.flink.cep.nfa.compiler.NFACompiler#compileFactory 构建 NFACompiler.NFAFactory

    org.apache.flink.cep.operator.CepOperator#open() 初始化逻辑

    1.  @Override
    2.     public void open() throws Exception {
    3.         super.open();
    4.         //TODO 初始化 定时器服务
    5.         timerService =
    6.                 getInternalTimerService(
    7.                         "watermark-callbacks", VoidNamespaceSerializer.INSTANCE, this);
    8.  
    9.         //TODO 创建NFA初始化了所有的顶点state和边transition
    10.         // 这个时候的state集合已经初始化完成了
    11.         nfa = nfaFactory.createNFA();
    12.         //TODO 给判断逻辑设置 cep运行环境
    13.         nfa.open(cepRuntimeContext, new Configuration());
    14.  
    15.         context = new ContextFunctionImpl();
    16.         collector = new TimestampedCollector<>(output);
    17.         cepTimerService = new TimerServiceImpl();
    18.  
    19.         // metrics
    20.         this.numLateRecordsDropped = metrics.counter(LATE_ELEMENTS_DROPPED_METRIC_NAME);
    21.     }

    org.apache.flink.cep.operator.CepOperator#processElement() 接收数据进行处理

    1.  @Override
    2.     public void processElement(StreamRecord<IN> element) throws Exception {
    3.         //TODO 判断当前语义为 ProcessingTime 还是 EventTime
    4.         if (isProcessingTime) {
    5.             if (comparator == null) {
    6.                 // there can be no out of order elements in processing time
    7.                 NFAState nfaState = getNFAState();
    8.                 long timestamp = getProcessingTimeService().getCurrentProcessingTime();
    9.                 advanceTime(nfaState, timestamp);
    10.                 processEvent(nfaState, element.getValue(), timestamp);
    11.                 updateNFA(nfaState);
    12.             } else {
    13.                 long currentTime = timerService.currentProcessingTime();
    14.                 bufferEvent(element.getValue(), currentTime);
    15.  
    16.                 // register a timer for the next millisecond to sort and emit buffered data
    17.                 timerService.registerProcessingTimeTimer(VoidNamespace.INSTANCE, currentTime + 1);
    18.             }
    19.  
    20.         } else {
    21.  
    22.             //TODO 获取事件的时间
    23.             long timestamp = element.getTimestamp();
    24.             //TODO 获取数据
    25.             IN value = element.getValue();
    26.  
    27.             // In event-time processing we assume correctness of the watermark.
    28.             // Events with timestamp smaller than or equal with the last seen watermark are
    29.             // considered late.
    30.             // Late events are put in a dedicated side output, if the user has specified one.
    31.             //TODO 在事件时间处理中，我们假设水印的正确性。
    32.             // 时间戳小于或等于最后看到的水印的事件被认为是迟到的。
    33.             // 如果用户指定了，则将延迟事件放在专用端输出中。
    34.             if (timestamp > timerService.currentWatermark()) {
    35.  
    36.                 // we have an event with a valid timestamp, so
    37.                 // we buffer it until we receive the proper watermark.
    38.  
    39.                 //TODO 注册Watermark定时器
    40.                 saveRegisterWatermarkTimer();
    41.  
    42.                 //TODO 缓存数据 key为事件时间 value为数据集合
    43.                 bufferEvent(value, timestamp);
    44.  
    45.             } else if (lateDataOutputTag != null) {
    46.                 output.collect(lateDataOutputTag, element);
    47.             } else {
    48.                 numLateRecordsDropped.inc();
    49.             }
    50.         }
    51.     }

    org.apache.flink.cep.operator.CepOperator#onEventTime() 触发计算

    1.  @Override
    2.     public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
    3.  
    4.         // 1) get the queue of pending elements for the key and the corresponding NFA,
    5.         // 2) process the pending elements in event time order and custom comparator if exists
    6.         //		by feeding them in the NFA
    7.         // 3) advance the time to the current watermark, so that expired patterns are discarded.
    8.         // 4) update the stored state for the key, by only storing the new NFA and MapState iff they
    9.         //		have state to be used later.
    10.         // 5) update the last seen watermark.
    11.  
    12.         //TODO
    13.         //  1)获取键的挂起元素队列和相应的NFA，
    14.         //  2)按照事件时间顺序处理挂起的元素，如果存在自定义比较器，则在NFA中输入它们
    15.         //  3)将时间提前到当前水印，丢弃过期的图案。
    16.         //  4)更新密钥的存储状态，只存储新的 NFA 和 MapState，如果它们有状态要稍后使用。
    17.         //  5)更新最后一次出现的水印。
    18.  
    19.         // STEP 1
    20.         //TODO 获取优先队列的中的数据
    21.         PriorityQueue<Long> sortedTimestamps = getSortedTimestamps();
    22.         //TODO 用于保存未匹配的完成的状态，和已匹配完成的状态，这里get为空时会初始化
    23.         // 先遍历所有的找到其中为start的state作为下一个可匹配的状态
    24.         NFAState nfaState = getNFAState();
    25.  
    26.         // STEP 2
    27.         while (!sortedTimestamps.isEmpty()
    28.                 && sortedTimestamps.peek() <= timerService.currentWatermark()) {
    29.             long timestamp = sortedTimestamps.poll();
    30.             //TODO 处理超时未匹配的数据
    31.             advanceTime(nfaState, timestamp);
    32.             try (Stream<IN> elements = sort(elementQueueState.get(timestamp))) {
    33.                 elements.forEachOrdered(
    34.                         event -> {
    35.                             try {
    36.                                 //TODO 将数据排好序（事件时间）以后使用NFA真正的处理逻辑
    37.                                 processEvent(nfaState, event, timestamp);
    38.                             } catch (Exception e) {
    39.                                 throw new RuntimeException(e);
    40.                             }
    41.                         });
    42.             }
    43.             elementQueueState.remove(timestamp);
    44.         }
    45.  
    46.         // STEP 3
    47.         advanceTime(nfaState, timerService.currentWatermark());
    48.  
    49.         // STEP 4
    50.         updateNFA(nfaState);
    51.  
    52.         if (!sortedTimestamps.isEmpty() || !partialMatches.isEmpty()) {
    53.             saveRegisterWatermarkTimer();
    54.         }
    55.     }

    org.apache.flink.cep.operator.CepOperator#processEvent() 处理数据

    1. /**
    2.      * Process the given event by giving it to the NFA and outputting the produced set of matched
    3.      * event sequences.
    4.      *
    5.      * @param nfaState Our NFAState object
    6.      * @param event The current event to be processed
    7.      * @param timestamp The timestamp of the event
    8.      */
    9.     private void processEvent(NFAState nfaState, IN event, long timestamp) throws Exception {
    10.         try (SharedBufferAccessor<IN> sharedBufferAccessor = partialMatches.getAccessor()) {
    11.             //TODO 得到匹配上规则的map，map中包含了这个正则的所有数据
    12.             Collection<Map<String, List<IN>>> patterns =
    13.                     nfa.process( //TODO 真正的处理逻辑
    14.                             sharedBufferAccessor,
    15.                             nfaState,
    16.                             event,
    17.                             timestamp,
    18.                             afterMatchSkipStrategy,
    19.                             cepTimerService);
    20.  
    21.             //TODO 这个map包含了匹配上的一个正则，下面会调用用户的select或者flatselect方法，往下游发送
    22.             processMatchedSequences(patterns, timestamp);
    23.         }
    24.     }

    org.apache.flink.cep.nfa.NFA#doProcess 开始处理数据

    ===> 第一条数据为 (1001,1656914303000,success)

    当前正在匹配的只有 begin 的state

    经过 org.apache.flink.cep.nfa.NFA#computeNextStates 匹配之后

    新的匹配中的state 变为两个 同时版本值会递增 同状态+1 下一个状态新增下一级

    org.apache.flink.cep.operator.CepOperator#processMatchedSequences 判断是否匹配完成输出

    ===> 第二条数据为 (1001,1656914304000,fail)

    经过 org.apache.flink.cep.nfa.NFA#computeNextStates 匹配之后

    ===> 第三数据为 (1001,1656914305000,fail)

    经过 org.apache.flink.cep.nfa.NFA#computeNextStates 匹配之后

    ===> 第四数据为(1001,1656914306000,success)

    经过 org.apache.flink.cep.nfa.NFA#computeNextStates 匹配之后

    此时当前数据匹配到begin，startState版本+1，并且新增一个middle的状态

    ===> 第五数据为(1001,1656914307000,end)

    经过 org.apache.flink.cep.nfa.NFA#computeNextStates 匹配

    正在匹配中的状态变成3个，两个end的state变为完成状态。

    判断匹配到state状态是否为Final状态，是添加到potentialMatches中。

    之后遍历从sharedBufferAccessor中获取数据。

    SharedBufferAccessor 简介

    Flink CEP设计了一个带版本的共享缓冲区。它会给每一次匹配分配一个版本号并使用该版本号来标记在这次匹配中的所有指针。

       

    做三个独立的缓冲区实现上是没有问题，但是我们发现缓冲区3状态stat1的堆栈和缓冲区1状态stat1的堆栈是一样的，我们完全没有必要分别占用内存。而且在实际的模式匹配场景下，每个缓冲区独立维护的堆栈中可能会有大量的数据重叠。随着流事件的不断流入，为每个匹配结果独立维护缓存区占用内存会越来越大。所以Flink CEP 提出了共享缓存区的概念(SharedBuffer),就是用一个共享的缓存区来表示上面三个缓存区。

    org.apache.flink.cep.operator.CepOperator#processMatchedSequences 处理匹配完成的state

    processMatchedSequences内部调用自定义的 select函数 输出满足模式的数据

    3.参考链接

    Flink源码解读系列 | Flink中的CEP复杂事件处理源码分析

    Flink CEP-NFA详解_听挽风讲大数据的博客-CSDN博客_flink nfa

  • 相关阅读:
    吐血整理，最全Pytest自动化测试框架快速上手（超详细）
    如何对用OpenCV开发的API进行测试 （Google Test 版本）
    工厂模式 与 抽象工厂模式 的区别
    德克萨斯大学奥斯汀分校自然语言处理硕士课程汉化版(第九周) - 可解释性
    艾美捷Abnova MYOC (人)抗体对说明书
    Spring框架
    文件上传漏洞
    python批量修改图像像素，修改命名，二值化，划分数据集
    线上环境内存溢出-OutOfMemoryError
    python监听html click教程
  • 原文地址：https://blog.csdn.net/qq_20672231/article/details/125615807