Watermark在Flink CEP中远远落后

Question

val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
streamEnv.addSource(new FlinkKafkaConsumer[..]())
          .filter(...)
          .map(...)
          .assignTimestampsAndWatermarks(
            WatermarkStrategy.forMonotonousTimestamps[Event]().withTimestampAssigner(..)
          )
          .keyBy(...)(TypeInformation.of(classOf[...]))
    
val pattern: Pattern[Event, _] = 
          Pattern.begin[Event]("start", AfterMatchSkipStrategy.skipPastLastEvent()).times(1)
          .next("middle")
          .oneOrMore()
          .optional()
          .where(new IterativeCondition[Event] {
             override def filter(event: Event, ctx:...): Boolean = {
                 val startTrafficEvent = ctx.getEventsForPattern("start").iterator().next()
                 startTrafficEvent.getFieldValue().equals(event.getFieldValue())
             }
          })
          .next("end").times(1)
          .where(new IterativeCondition[Event] {
             override def filter(event: Event, ctx:...): Boolean = {
                  val startTrafficEvent = ctx.getEventsForPattern("start").iterator().next()
                  !startTrafficEvent.getFieldValue().equals(event.getFieldValue())
            }
          })
          .within(Time.seconds(30))

在Web UI中，有两个任务：第一个任务是源->过滤器->映射->时间戳/水印；第二个是cepoperator->sink。每个任务得到104个并行度。

子任务的工作负载不均匀，它应该来自keyby。子任务之间的水印是不同的，但是它们开始被固定在一个值上，在很长一段时间内没有变化。从日志中，我可以看到CEP不断评估事件，并将匹配的结果推送到下游sink。

事件率为10k/s,第一个任务的背压保持“高码”,第二个任务的背压保持“好码”,第一个任务的背压保持“高码”,第二个任务的背压保持“好码”,第二个任务的背压保持“好码”。

谢谢

Answer 1

对你的问题作了更仔细的考虑后，我正在修改我的回答。

听起来CEP正在继续产生火柴，它们正在被推到水槽，但是CEP+水槽任务正在产生很高的背压。找出背压的原因会有帮助。

如果事件可以从所有分区读取，而水印只是勉强向前推进，听起来背压很大，足以阻止事件被吸收。

null

以下是一些获得更多洞察力的想法：

（1）尝试使用探查器来确定CepOperator是否是瓶颈，并可能识别它正在做什么。

（2）禁用CepOperator和sink之间的操作符链接，以便隔离CEP--仅作为调试步骤。这将为您提供更好的可见性（通过度量和背压监控），以了解CEP和水槽各自在做什么。